نظریهٔ سیستم‌های توزیع‌شدهDistributed Systems Theory

Let me start with a hard truth that separates a mid-level engineer from a real senior: on a single machine, either everything works or everything fails. In a distributed system, something is always broken and you don't know which part. This is called partial failure, and nearly all of the difficulty in this field flows from that one sentence.

When you call save() on a local database, you get two outcomes: success or error. But when that request travels over the network to another service, a third, devilish outcome appears: you don't know. Maybe the request never arrived. Maybe it arrived, was processed, but the reply was lost. Maybe it's still processing and will answer in three seconds. That "I don't know" is the beating heart of distributed systems theory.

This chapter is not going to parrot academic formulas. After reading it, when someone in an interview or a design review asks "OK, so what happens here if the network partitions?", you should answer not with fear but with a precise mental map: knowing what is theoretically impossible, what trade-off you're making, and which pattern solves your pain.

1) Why is distributed computing so hard?

The network lies to you

In 1994, Peter Deutsch and colleagues at Sun compiled a list called "The Eight Fallacies of Distributed Computing." These are false assumptions that every junior engineer unconsciously makes, and then pays for in production with blood and bone:

1. The network is reliable. 2) Latency is zero. 3) Bandwidth is infinite. 4) The network is secure. 5) Topology doesn't change. 6) There is one administrator. 7) Transport cost is zero. 8) The network is homogeneous.

Why a timeout is not a definitive answer

A subtle point most people miss: a timeout never tells you whether the other side is dead or just slow. If you waited 5 seconds and got no reply, you have three indistinguishable possibilities: (a) the server died, (b) the server is alive but slow and will answer in 6 seconds, (c) the answer was ready but got lost on the way back. This uncertainty makes decisions hard: if you assume it's dead and resend, the work might be done twice (a duplicate). If you assume it's alive and wait, you might wait forever.

2) Failure models: know your enemy precisely

Before you build a "resilient" system, you must define precisely: resilient against what kind of failure? Failure models range from "gentle" to "evil":

A crucial point: a timing failure is more dangerous than a crash failure, because a slow node is worse than a dead node. You remove a dead node from the loop and you're done. But a slow node shows green in the health check, still takes traffic, and slows everyone down. This phenomenon is called "gray failure" and is the cause of many real incidents.

3) Time and ordering: why you can't trust the clock

The wall clock lies

The biggest beginner mistake in a distributed system is thinking that System.currentTimeMillis() is comparable across two machines. It isn't.

// Wrong: with NTP this can go negative
long start = System.currentTimeMillis();
doWork();
long elapsed = System.currentTimeMillis() - start; // may be wrong!

// Right: the monotonic clock only moves forward
long startNanos = System.nanoTime();
doWork();
long elapsedNanos = System.nanoTime() - startNanos; // always valid

The problem is deeper: even with NTP, two servers' clocks usually differ by a few milliseconds (and in the worst case a few seconds) — this is clock skew. So the logic "the record with the newer timestamp wins" (last-write-wins) is inherently buggy: a write that really happened later might carry a smaller timestamp and be wrongly discarded.

happens-before: causal order instead of clock order

In his legendary 1978 paper, Leslie Lamport said: forget "what time" it happened. Just ask "could A have influenced B?". We call this relation "happens-before" and denote it $\rightarrow$:

• If A and B are in the same process and A ran earlier: $A \rightarrow B$.
• If A is sending a message and B is receiving that message: $A \rightarrow B$.
• It's transitive: if $A \rightarrow B$ and $B \rightarrow C$ then $A \rightarrow C$.

If neither $A \rightarrow B$ nor $B \rightarrow A$, they are concurrent — meaning they have no causal relation and their order is meaningless.

The diagram below shows the happens-before relation: sending a message creates a causal edge. | این نمودار رابطهٔ happens-before را نشان می‌دهد: فرستادن پیام یک لبهٔ علّی می‌سازد.

sequenceDiagram
    participant P1 as Process 1
    participant P2 as Process 2
    P1->>P1: event a
    P1->>P2: message m (event b = send)
    Note over P2: event c = receive m
    P2->>P2: event d
    Note over P1,P2: a -> b -> c -> d (causal chain)

Lamport clock: a logical number per event

A Lamport clock is a simple counter on each process that translates this causal relation into a number. Its rule is three lines:

public class LamportClock {
    private long counter = 0;

    // before each local event: increment
    public synchronized long tick() {
        return ++counter;
    }

    // when sending a message: tick and send the number with the message
    public synchronized long onSend() {
        return ++counter;
    }

    // on receive: max of self and sender, then increment
    public synchronized long onReceive(long senderClock) {
        counter = Math.max(counter, senderClock) + 1;
        return counter;
    }
}

Lamport's guarantee: if $A \rightarrow B$ then $L(A) < L(B)$. But the reverse is not true: if $L(A) < L(B)$ it does not necessarily mean A caused B; they might just be concurrent. That's the fundamental limitation of Lamport clocks.

Vector clock: detecting concurrency

A vector clock fixes this weakness. Instead of one number, each node keeps an array: one slot per node. It increments its own slot for each event, and on receive it takes an element-wise maximum.

public class VectorClock {
    private final int nodeId;
    private final long[] vector;

    public VectorClock(int nodeId, int numNodes) {
        this.nodeId = nodeId;
        this.vector = new long[numNodes];
    }

    public synchronized long[] tick() {
        vector[nodeId]++;
        return vector.clone();
    }

    public synchronized void onReceive(long[] other) {
        for (int i = 0; i < vector.length; i++) {
            vector[i] = Math.max(vector[i], other[i]);
        }
        vector[nodeId]++;
    }

    // A is before B if every slot is <= and at least one is <
    public static boolean happensBefore(long[] a, long[] b) {
        boolean strictlyLess = false;
        for (int i = 0; i < a.length; i++) {
            if (a[i] > b[i]) return false;
            if (a[i] < b[i]) strictlyLess = true;
        }
        return strictlyLess;
    }
}

Now if neither $A \rightarrow B$ nor $B \rightarrow A$, you know with certainty they were concurrent — and that's exactly what an eventually consistent database like early Riak/DynamoDB needs to detect a conflict. When two concurrent writes hit the same key, the system keeps both as "siblings" and asks the application to resolve the conflict.

4) Consensus: the hardest problem in the field

Consensus means several nodes agreeing on one value, even if some nodes fail. It's the foundation of everything: leader election, committing a transaction, recording log order. We want three properties: (1) Agreement: everyone agrees on one value. (2) Validity: the agreed value must be one of the proposed values. (3) Termination: a decision is eventually made.

Paxos at a glance

Paxos (also Lamport's work) was the first proven consensus algorithm and sits at the heart of systems like Google Chubby and Spanner. It has three roles: Proposer, Acceptor, Learner. It works in two phases: phase 1 (Prepare/Promise), where a proposer obtains permission with a unique number, and phase 2 (Accept/Accepted), where the value gets locked in. A majority of acceptors suffices. Paxos is famous for being hard to understand; Lamport himself had to write "Paxos Made Simple," and even then people struggle to implement it. Multi-Paxos is its practical version for a sequence of decisions.

Raft: understandable consensus

Raft was created in 2014 by Ongaro and Ousterhout with an explicit goal: understandability. Today it's the consensus engine in etcd (and therefore Kubernetes), Consul, RabbitMQ (quorum queues), ClickHouse, and Kafka's KRaft mode. Raft decomposes the problem into three independent subproblems: leader election, log replication, and safety.

Each node is in one of three states and moves between them based on the term (a logical time counter that plays the role of a "presidential term").

The state diagram below shows the lifecycle of a Raft node. | نمودار حالت زیر چرخهٔ عمرِ یک نودِ Raft را نشان می‌دهد.

stateDiagram-v2
    [*] --> Follower
    Follower --> Candidate: election timeout\n(no heartbeat)
    Candidate --> Candidate: split vote\n(new term, retry)
    Candidate --> Leader: wins majority
    Candidate --> Follower: sees higher term\nor new leader
    Leader --> Follower: discovers higher term
    Leader --> Leader: send heartbeats

Leader election: each follower has a randomized election timeout (typically 150 to 300 ms). If it receives no heartbeat from the leader within that window, it assumes the leader died: it increments its own term, becomes a candidate, votes for itself, and requests votes (RequestVote) from the others. Each node votes only once per term. If a candidate gets a majority, it becomes leader and starts sending heartbeats. The randomized timeout is the key: it makes two nodes rarely become candidates simultaneously and reduces split votes — this is exactly where Raft sidesteps FLP.

Log replication: only the leader accepts writes from clients. It appends each command as an entry to its log and sends it in AppendEntries to followers. Once a majority of nodes have written the entry to disk, the leader marks it "committed," applies it to the state machine, and replies to the client. This "majority write before commit" guarantees that even if the leader immediately fails, any new leader (which must have a majority vote) definitely has that entry.

5) Replication and quorums: the golden R+W>N rule

When you copy data across $N$ replicas, you choose two numbers yourself: for each write, you wait for acks from $W$ replicas; for each read, you ask $R$ replicas and take the newest. This Dynamo-style model (used by Cassandra, Riak, and ScyllaDB) lets you tune consistency per request.

For example with $N=3$: choosing $W=2, R=2$ is balanced ($2+2>3$). If you want fast reads, $W=3, R=1$ (slow writes, instant reads). If you want fast writes, $W=1, R=3$. And $W=1, R=1$ means maximum speed but no guarantee.

To compensate for inconsistency, these systems have two background mechanisms: read repair (when a read sees different versions, it pushes the newest back to the lagging nodes) and anti-entropy with a Merkle tree, which I'll explain in the gossip section.

6) CAP and PACELC: the fundamental decision

CAP: the forced choice at partition time

The CAP theorem (proposed by Eric Brewer, proven by Gilbert and Lynch) names three properties: Consistency (everyone sees one value), Availability (every request gets a response), Partition tolerance (the system works despite a network cut).

The industry's biggest misconception is thinking "pick 2 of 3." That's wrong. The truth is subtler:

• CP (like etcd, ZooKeeper, HBase, MongoDB with majority): during a partition, the minority side refuses to respond so it doesn't serve contradictory data. It preserves consistency, sacrifices availability.
• AP (like Cassandra, DynamoDB, Riak): both sides of the partition respond, even if the data is stale; they reconcile later (eventual). Preserves availability, temporarily sacrifices consistency.

PACELC: what CAP left out

The CAP theorem has a big flaw: it only talks about partition time, which rarely happens. But your system is in normal state 99.9% of the time — and there too you have a trade-off. Daniel Abadi completed the picture in 2012 with PACELC:

7) The consistency spectrum: from strong to weak

"Consistency" is a vague word; it's really a spectrum. From strongest (most expensive) to weakest:

• Linearizable (atomic): the strongest. The system behaves as if there is a single copy of the data and every operation takes effect at an atomic instant between its start and end. If my write completed, every subsequent read (by any clock) definitely sees it. This is what a local variable gives you. The most expensive, because it needs global coordination.
• Sequential: everyone sees all operations in one single total order, and each client's operation order is preserved — but that order need not match real wall-clock time.
• Causal: only causally-related operations (happens-before) are seen in order; concurrent operations may be seen in any order. This is the "sweet spot": the strongest consistency achievable without sacrificing availability.
• Read-your-writes: guarantees that after I write something, I myself see it (though others may not yet). Many UX bugs come from lacking exactly this.
• Monotonic reads: if I've seen a value once, my later reads won't go backward.
• Eventual: the weakest. It only promises "if writes stop, all replicas eventually converge." It says nothing about "when."

The diagram below orders the consistency spectrum from strong to weak. | نمودار زیر طیفِ سازگاری را از قوی به ضعیف مرتب می‌کند.

flowchart TD
    L[Linearizable\nstrongest, costliest] --> S[Sequential]
    S --> C[Causal\nbest without losing availability]
    C --> RYW[Read-your-writes]
    RYW --> MR[Monotonic reads]
    MR --> E[Eventual\nweakest, cheapest]

8) Message delivery: why "exactly-once" is a myth

When A sends a message to B and the message might be lost, you have three possible guarantees:

• At-most-once: fire and forget; if it's lost, it's lost. You get no duplicates but might lose messages. Good for metrics and telemetry.
• At-least-once: retry until you get an ack. A message is never lost, but you will definitely get duplicates someday (because the returning ack might be lost and you resend). This is the realistic default for most systems.
• Exactly-once: everyone's wish; each message takes effect exactly once.

Idempotency: the main weapon

An idempotent operation means running it again yields the same result as running it once. balance = 100 (set) is idempotent; balance += 10 (increment) is not. The standard pattern is to give each message a unique key (idempotency key) and check before processing whether you've seen it before:

@Service
public class PaymentConsumer {

    private final ProcessedMessageRepository processed;
    private final PaymentService payments;

    @KafkaListener(topics = "payments")
    @Transactional
    public void onMessage(PaymentEvent event) {
        // dedup: if we've seen this messageId before, skip
        if (processed.existsById(event.getMessageId())) {
            return; // duplicate — safely ignore
        }
        payments.charge(event.getAccountId(), event.getAmount());
        // record dedup and effect in one atomic transaction
        processed.save(new ProcessedMessage(event.getMessageId()));
    }
}

A crucial point: recording the dedup key and the business effect must be in one atomic transaction, otherwise there's a window where the work is done but dedup isn't recorded (or vice versa). This is the inbox pattern. The dedup table looks like this — and its upsert differs between Oracle and PostgreSQL:

-- PostgreSQL: ON CONFLICT for an idempotent insert
INSERT INTO processed_message (message_id, processed_at)
VALUES (:id, now())
ON CONFLICT (message_id) DO NOTHING;

-- Oracle (19c/23ai): MERGE for the same logic
MERGE INTO processed_message t
USING (SELECT :id AS message_id FROM dual) s
ON (t.message_id = s.message_id)
WHEN NOT MATCHED THEN
  INSERT (message_id, processed_at) VALUES (s.message_id, SYSTIMESTAMP);

Exactly-once in Kafka

Kafka builds "exactly-once semantics" from version 0.11 with two mechanisms. First, the idempotent producer: each producer gets a PID and each record a sequence number; using these two, the broker deduplicates repeated retries on the log. Second, transactions: it makes writing across multiple partitions and committing consumer offsets atomic.

9) Distributed transactions: 2PC vs saga

When one operation must be atomic across several services/databases (all or nothing), you have two schools.

Two-Phase Commit (2PC)

A coordinator runs two phases. Phase 1 (prepare): it asks everyone "are you ready to commit?" and each participant takes a lock and says "yes/no." Phase 2 (commit): if everyone said yes, the coordinator sends commit to all; otherwise abort.

The diagram below shows a successful 2PC. | نمودار زیر یک 2PCِ موفق را نشان می‌دهد.

sequenceDiagram
    participant C as Coordinator
    participant A as Service A
    participant B as Service B
    C->>A: prepare
    C->>B: prepare
    A-->>C: yes (locked)
    B-->>C: yes (locked)
    C->>A: commit
    C->>B: commit
    A-->>C: ack
    B-->>C: ack

Saga: the practical distributed transaction

The saga pattern breaks a large transaction into a sequence of small local transactions, each committed on one service. If a step fails, instead of a rollback (impossible in a distributed system), you run compensating transactions in reverse order to undo the effects of previous steps.

It has two styles: orchestration (a central coordinator commands the steps; readable and traceable) and choreography (each service reacts to events with events; decoupled but harder to trace the flow).

The diagram below shows an order saga with a failure and compensation. | نمودار زیر یک saga سفارش با یک شکست و جبران را نشان می‌دهد.

sequenceDiagram
    participant O as Order
    participant P as Payment
    participant I as Inventory
    O->>P: reserve payment
    P-->>O: paid
    O->>I: reserve stock
    I-->>O: OUT OF STOCK (fail)
    Note over O,P: compensate in reverse
    O->>P: refund payment
    P-->>O: refunded

10) Gossip and anti-entropy: spreading information like a rumor

When you have hundreds of nodes, you can't broadcast every change from one central point to everyone (single point of failure and bottleneck). Instead you use gossip (an epidemic protocol): every second each node talks to a few random nodes and exchanges its information. Like a spreading rumor, information propagates exponentially and fault-tolerantly across the whole cluster. Cassandra, Consul, and Serf learn cluster membership and health status via gossip.

Anti-entropy is a background process that finds and repairs data differences between replicas. Since comparing all the data is expensive, they use a Merkle tree: a hash tree whose leaves are hashes of data blocks and whose higher nodes are hashes of their children. Two nodes compare only the root; if they match, there's no difference and it's done. If they differ, they walk down the tree and sync only the branches whose hashes differ — instead of transferring the whole dataset, only the differences move.

11) CRDT: data that resolves conflicts itself

Suppose two users edit a shopping cart offline, then both sync. Who wins? With last-write-wins, one of the changes is lost. A CRDT (Conflict-free Replicated Data Type) is a special kind of data structure designed so that merging two versions always gives a deterministic, conflict-free result without any coordination. It's the basis of local-first and collaborative apps like Redis (CRDT type in Active-Active), Riak, and shared editors.

It has two families. State-based (CvRDT): each node sends its entire state and a merge function combines them; this merge must be commutative, associative, and idempotent so that message order and repetition don't matter (a join-semilattice). Operation-based (CmRDT): it only broadcasts the operations themselves, which are lighter but require more precise delivery.

The simplest example is the G-Counter (grow-only counter): each node only increments its own slot, and merge is an element-wise maximum:

// G-Counter: distributed grow-only counter, conflict-free merge
public class GCounter {
    private final int nodeId;
    private final long[] counts;

    public GCounter(int nodeId, int numNodes) {
        this.nodeId = nodeId;
        this.counts = new long[numNodes];
    }

    public void increment() { counts[nodeId]++; }

    public long value() {
        long sum = 0;
        for (long c : counts) sum += c;
        return sum;
    }

    // merge: element-wise max — commutative, associative, idempotent
    public void merge(GCounter other) {
        for (int i = 0; i < counts.length; i++) {
            counts[i] = Math.max(counts[i], other.counts[i]);
        }
    }
}

Because merge is just max, any order and any repetition of merges reaches the same answer — exactly the property you want for eventual consistency without coordination. For a decrementable counter there's the PN-Counter (two G-Counters, one for + and one for -), for a set there's the OR-Set (which manages add and remove with a unique tag), and for a single value there's the LWW-Register.

12) Split-brain and fencing tokens: the most dangerous distributed bug

Split-brain means that due to a network partition, two nodes simultaneously think they're the leader and both start writing. The result: corrupted data, two primaries overwriting each other, money spent twice. This is the nightmare of every distributed system.

The correct fix is a fencing token: every time a lock is granted, a monotonically increasing number is issued too. Every write must carry its token, and the data source (database/storage) rejects any write whose token is smaller than the last token it saw.

The diagram below shows how a fencing token stops a zombie node. | نمودار زیر نشان می‌دهد چطور fencing token جلوی نودِ زامبی را می‌گیرد.

sequenceDiagram
    participant A as Node A (paused)
    participant L as Lock Service
    participant B as Node B
    participant S as Storage
    A->>L: acquire lock -> token 33
    Note over A: long GC pause...
    L->>B: lease expired, grant -> token 34
    B->>S: write with token 34 (accepted)
    A->>S: write with token 33 (STALE)
    S-->>A: REJECTED (33 < 34)

// storage accepts only writes with a token >= the last one
public class FencedStorage {
    private long lastToken = 0;

    public synchronized void write(long fencingToken, byte[] data) {
        if (fencingToken < lastToken) {
            throw new StaleTokenException(
                "token " + fencingToken + " < " + lastToken + " — rejected");
        }
        lastToken = fencingToken;
        persist(data);
    }
}

Because the token is monotonically increasing, zombie node A with the old token (33) can never overwrite the live node B's write (34). Real systems: ZooKeeper with zxid, etcd with revision, and version-based locks all provide this number.