10 - Distributed Systems and Concurrency: CAP Theorem, Consistency Models and CRDT

When concurrency crosses the network

So far the series has dealt with threads, tasks and locks inside a single process, where memory is shared and a message is not lost. As soon as data lives on several nodes, those guarantees have to be earned: a message can arrive late or never, a node can restart, two nodes may not see each other even though both are alive. A concrete case: a shopping cart replicated across three servers, and a customer who adds an item from their phone just while one server is cut off from the others. Which server does the site answer from, and what does the customer see on the next request?

The answers have names: CAP, consistency models, CRDTs. The examples here are simulations in Python (3.14.4, standard library only, all run): they simplify the network to a set of nodes that see each other or not, and do not replace a test on a real database. For coordination inside a process, the thread of the series is in race conditions and deadlocks.

CAP: what it really says

The CAP theorem concerns three properties of a replicated system:

• Consistency (C): every read sees the latest write, or receives an error.
• Availability (A): every request received by a working node gets a response.
• Partition tolerance (P): the system keeps working even when the network loses or delays messages between nodes.

The most common mistake is to read "pick two out of three" as a menu. A network partition is not chosen: it happens. So, when it happens, the system must decide between C and A, and that is all there is to it. Daniel Abadi, in the paper proposing PACELC, says it in precise words: CAP imposes restrictions only when there is a partition and does not limit the system's capabilities in normal operation. My reading is: C and A are in conflict only during the partition; outside the partition there remains another trade-off, the one between latency and consistency, which CAP does not describe.

To see the choice, the simulation below has three replicas A, B and C, and a partition separating {A, B} from {C}. The CP system requires a majority to write and to read; the AP system always writes locally and reconciles afterwards with "last write wins" (LWW, based on a timestamp).

from dataclasses import dataclass, field

class Indisponibile(Exception):
    pass

@dataclass
class Replica:
    nome: str
    valore: str = "v0"
    ts: int = 0                      # logical timestamp of the write

@dataclass
class Cluster:
    repliche: dict
    partizione: list = field(default_factory=list)   # groups of nodes that can see each other

    def raggiungibili(self, da: str) -> list:
        if not self.partizione:
            return list(self.repliche)
        return next(g for g in self.partizione if da in g)

    # CP: the write requires the MAJORITY, otherwise it refuses
    def scrivi_cp(self, da: str, valore: str, ts: int) -> str:
        gruppo = self.raggiungibili(da)
        if len(gruppo) <= len(self.repliche) // 2:
            raise Indisponibile(f"{da} sees only {len(gruppo)} out of {len(self.repliche)}: write rejected")
        for n in gruppo:
            self.repliche[n].valore, self.repliche[n].ts = valore, ts
        return f"{da}: written '{valore}' on {len(gruppo)} nodes"

    # CP: the read also requires the majority, otherwise it refuses
    def leggi_cp(self, da: str) -> str:
        gruppo = self.raggiungibili(da)
        if len(gruppo) <= len(self.repliche) // 2:
            raise Indisponibile(f"{da} sees only {len(gruppo)} out of {len(self.repliche)}: read rejected")
        return max((self.repliche[n] for n in gruppo), key=lambda r: r.ts).valore

    # AP: ALWAYS writes locally (and on the reachable nodes), reconciles later
    def scrivi_ap(self, da: str, valore: str, ts: int) -> str:
        for n in self.raggiungibili(da):
            self.repliche[n].valore, self.repliche[n].ts = valore, ts
        return f"{da}: written '{valore}' on {len(self.raggiungibili(da))} nodes"

    def ripara_lww(self) -> None:           # last-write-wins: the highest timestamp wins
        vincitore = max(self.repliche.values(), key=lambda r: (r.ts, r.nome))
        for r in self.repliche.values():
            r.valore, r.ts = vincitore.valore, vincitore.ts

def nuovo() -> Cluster:
    c = Cluster({n: Replica(n) for n in "ABC"})
    c.partizione = [["A", "B"], ["C"]]       # the network separates {A,B} from {C}
    return c

print("--- CP with partition {A,B} | {C}")
cp = nuovo()
print(cp.scrivi_cp("A", "da-A", ts=1))
try:
    cp.scrivi_cp("C", "da-C", ts=2)
except Indisponibile as e:
    print("C:", e)
print("read on A:", cp.leggi_cp("A"))
try:
    cp.leggi_cp("C")
except Indisponibile as e:
    print("C:", e)

print("--- AP with the same partition")
ap = nuovo()
print(ap.scrivi_ap("A", "da-A", ts=1))
print(ap.scrivi_ap("C", "da-C", ts=2))
print("before the repair:", {n: r.valore for n, r in ap.repliche.items()})
ap.partizione = []
ap.ripara_lww()
print("after the LWW repair:", {n: r.valore for n, r in ap.repliche.items()}, "-> the write 'da-A' was lost")

# Output:
# --- CP with partition {A,B} | {C}
# A: written 'da-A' on 2 nodes
# C: C sees only 1 out of 3: write rejected
# read on A: da-A
# C: C sees only 1 out of 3: read rejected
# --- AP with the same partition
# A: written 'da-A' on 2 nodes
# C: written 'da-C' on 1 nodes
# before the repair: {'A': 'da-A', 'B': 'da-A', 'C': 'da-C'}
# after the LWW repair: {'A': 'da-C', 'B': 'da-C', 'C': 'da-C'} -> the write 'da-A' was lost

In the CP system, A writes to two nodes out of three (the majority) and can also read; C, which sees only itself, refuses both the write and the read: it is unavailable, but it can never return an old value. In the AP system both sides accept the write; on reunion the LWW rule picks the highest timestamp and da-A disappears without a trace. The cost of AP is not "a stale value" but a lost update if the reconciliation is not done with care: that is why vector clocks and CRDTs are needed, two sections further down.

Quorum: when "majority" really means something

One way to tune consistency is to say how many replicas must respond to a write (W) and to a read (R), out of N in total. The Cassandra documentation presents it as Dynamo's mechanism: if R + W > N, writes are visible to later reads because the two sets of replicas overlap. The program below checks it by enumeration, trying every possible pair of replica sets.

from itertools import combinations

def intersezione_garantita(n: int, w: int, r: int) -> bool:
    """Does every set of W replicas (write) meet every set of R replicas (read)?"""
    repliche = range(n)
    return all(set(a) & set(b)
               for a in combinations(repliche, w)
               for b in combinations(repliche, r))

for n, w, r in [(3, 2, 2), (3, 1, 1), (3, 3, 1), (3, 1, 3), (5, 3, 3), (5, 2, 3)]:
    print(f"N={n} W={w} R={r}  W+R>N: {w + r > n!s:<5}  every read meets the latest write: {intersezione_garantita(n, w, r)}")

# Output:
# N=3 W=2 R=2  W+R>N: True   every read meets the latest write: True
# N=3 W=1 R=1  W+R>N: False  every read meets the latest write: False
# N=3 W=3 R=1  W+R>N: True   every read meets the latest write: True
# N=3 W=1 R=3  W+R>N: True   every read meets the latest write: True
# N=5 W=3 R=3  W+R>N: True   every read meets the latest write: True
# N=5 W=2 R=3  W+R>N: False  every read meets the latest write: False

With N=3, W=2, R=2 every write and every read share at least one replica; with W=1, R=1 they do not, and a read may miss the latest write. W=3, R=1 and W=1, R=3 also work, with opposite costs: in the first the write waits for all the replicas, in the second it is the read. Beware of overestimating the result: Abadi's paper observes that increasing R + W gives more consistency at the price of more latency, but that full consistency in the sense of Gilbert and Lynch (linearizability) is not obtained even with R + W > N. Quorum intersection is a necessary condition, not a sufficient one, for guaranteeing that reads always see the latest write.

A menu of consistencies, from strongest to weakest

Between "consistent" and "inconsistent" there is a spectrum. Linearizability: every operation seems to happen at an instant between the request and the response, and everyone sees the same order; it is the strongest model and requires coordination. Sequential consistency: everyone agrees on a single order of operations, but it need not respect real time. Causal consistency: operations linked by cause and effect are seen in the same order by everyone, independent ones in any order. Eventual consistency: if writes stop, the replicas converge, with no guarantee about when. I do not report here which products offer which model: it depends on the configuration, and the rankings around are often approximate.

Causal consistency is the level that interests the next tool: to offer it, the system must be able to say whether two events are ordered or concurrent.

Vector clocks: recognizing concurrency

The system clock is not enough: two nodes have different clocks, and LWW has just shown it. Lamport, in his July 1978 paper on logical clocks, shows how to order events with counters incremented at every message; that construction gives an order, but does not let you recognize two concurrent events. Vector clocks (one counter per node) are a later evolution of that idea, and they are the ones needed. Attributing them to Lamport 1978, as is often read, is a mistake: the 1978 paper is about scalar logical clocks.

from dataclasses import dataclass, field

@dataclass(frozen=True)
class VectorClock:
    clocks: dict = field(default_factory=dict)

    def incrementa(self, nodo: str) -> "VectorClock":
        nuovo = dict(self.clocks)
        nuovo[nodo] = nuovo.get(nodo, 0) + 1
        return VectorClock(nuovo)

    def unisci(self, altro: "VectorClock") -> "VectorClock":
        chiavi = set(self.clocks) | set(altro.clocks)
        return VectorClock({k: max(self.clocks.get(k, 0), altro.clocks.get(k, 0)) for k in chiavi})

    def precede(self, altro: "VectorClock") -> bool:
        chiavi = set(self.clocks) | set(altro.clocks)
        return (all(self.clocks.get(k, 0) <= altro.clocks.get(k, 0) for k in chiavi)
                and any(self.clocks.get(k, 0) < altro.clocks.get(k, 0) for k in chiavi))

    def concorrente(self, altro: "VectorClock") -> bool:
        return not self.precede(altro) and not altro.precede(self)

base = VectorClock()
a1 = base.incrementa("A")                 # A modifies the cart
b1 = a1.incrementa("B")                   # B received a1, then modifies: causally AFTER a1
c1 = a1.incrementa("C")                   # C, which has seen only a1, modifies without knowing about B

print("a1", a1.clocks, "precedes b1", b1.clocks, ":", a1.precede(b1))
print("b1 and c1 are concurrent:", b1.concorrente(c1))
print("a1 and c1 are concurrent:", a1.concorrente(c1), "(a1 precedes c1:", a1.precede(c1), ")")
print("union b1+c1:", b1.unisci(c1).clocks)

# Output:
# a1 {'A': 1} precedes b1 {'A': 1, 'B': 1} : True
# b1 and c1 are concurrent: True
# a1 and c1 are concurrent: False (a1 precedes c1: True )
# union b1+c1: {'C': 1, 'A': 1, 'B': 1}

precede compares two vectors: A precedes B if every counter of A is less than or equal to that of B and at least one is strictly less. b1 derives from a1 (B had received it), so a1 precedes it; b1 and c1, on the other hand, are two independent changes to the same state, and neither precedes the other: they are concurrent. In an AP system this is exactly the signal needed: instead of overwriting like LWW, the system knows it must keep both versions and ask for them to be merged (by the application, or by a CRDT). The union of the two vectors takes the maximum counter per node.

CRDTs: data structures that merge themselves

A CRDT (conflict-free replicated data type) is a data structure designed to be replicated: each replica updates itself, without coordinating, and the replicas exchange state; the merge always leads to the same result. Preguiça, Baquero and Shapiro summarize the condition for state-based CRDTs: states form a join semilattice, every operation produces a state greater than or equal, and the merge is the least upper bound of the two states. In plain words: the merge must be commutative, associative and idempotent, so that the order and repetitions of messages do not matter.

The listing implements three CRDTs (G-Counter, PN-Counter, OR-Set) and checks the three properties on 2,000 triples of random states for two of them, then reproduces the most instructive OR-Set case: a concurrent add and remove of the same element.

import random
from dataclasses import dataclass, field

@dataclass(frozen=True)
class GCounter:
    """Growing counter: one partial counter per node."""
    conteggi: tuple = ()                       # (node, value) pairs, sorted

    def _d(self): return dict(self.conteggi)

    def incrementa(self, nodo: str, n: int = 1) -> "GCounter":
        d = self._d(); d[nodo] = d.get(nodo, 0) + n
        return GCounter(tuple(sorted(d.items())))

    def valore(self) -> int: return sum(v for _, v in self.conteggi)

    def unisci(self, altro: "GCounter") -> "GCounter":
        a, b = self._d(), altro._d()
        return GCounter(tuple(sorted((k, max(a.get(k, 0), b.get(k, 0))) for k in a.keys() | b.keys())))

@dataclass(frozen=True)
class PNCounter:
    """Counter that grows and shrinks: two G-Counters."""
    pos: GCounter = GCounter()
    neg: GCounter = GCounter()

    def incrementa(self, nodo): return PNCounter(self.pos.incrementa(nodo), self.neg)
    def decrementa(self, nodo): return PNCounter(self.pos, self.neg.incrementa(nodo))
    def valore(self): return self.pos.valore() - self.neg.valore()
    def unisci(self, altro): return PNCounter(self.pos.unisci(altro.pos), self.neg.unisci(altro.neg))

@dataclass(frozen=True)
class ORSet:
    """Observed-Remove Set: every add carries a unique tag; remove deletes only the tags it has SEEN."""
    aggiunte: frozenset = frozenset()          # (element, tag) pairs
    rimosse: frozenset = frozenset()           # deleted (element, tag) pairs

    def add(self, elemento, tag): return ORSet(self.aggiunte | {(elemento, tag)}, self.rimosse)
    def remove(self, elemento):
        visti = {(e, t) for (e, t) in self.aggiunte if e == elemento}
        return ORSet(self.aggiunte, self.rimosse | visti)
    def contiene(self, elemento): return any(e == elemento for (e, t) in self.aggiunte - self.rimosse)
    def unisci(self, altro): return ORSet(self.aggiunte | altro.aggiunte, self.rimosse | altro.rimosse)

# --- Merge properties, verified on 2000 random triples of states ---------------------------
random.seed(1)
def g_casuale():
    g = GCounter()
    for _ in range(random.randint(0, 6)): g = g.incrementa(random.choice("ABC"), random.randint(1, 5))
    return g

def or_casuale():
    s = ORSet()
    for i in range(random.randint(0, 6)):
        s = s.add(random.choice("xyz"), f"t{random.randint(0, 9)}")
        if random.random() < 0.3: s = s.remove(random.choice("xyz"))
    return s

for nome, gen in [("G-Counter", g_casuale), ("OR-Set", or_casuale)]:
    comm = assoc = idem = 0
    for _ in range(2000):
        a, b, c = gen(), gen(), gen()
        comm += a.unisci(b) == b.unisci(a)
        assoc += a.unisci(b).unisci(c) == a.unisci(b.unisci(c))
        idem += a.unisci(a) == a
    print(f"{nome:<10} commutative {comm}/2000, associative {assoc}/2000, idempotent {idem}/2000")

# --- Convergence of a PN-Counter: three replicas, isolated updates, merges in different orders --
a = PNCounter().incrementa("A").incrementa("A")        # A: +2
b = PNCounter().incrementa("B").decrementa("B")         # B: +1 -1
c = PNCounter().decrementa("C")                         # C: -1
x = a.unisci(b).unisci(c)
y = c.unisci(a).unisci(b).unisci(a)                     # different order, with a repeated merge
print("PN-Counter: value", x.valore(), "and", y.valore(), "- same result:", x == y)

# --- OR-Set: concurrent add and remove of the same element ----------------------------------------
base = ORSet().add("milk", "A1")
replica_a = base.remove("milk")                         # A removes the milk it has seen (tag A1)
replica_b = base.add("milk", "B1")                      # B, unknowingly, adds it again (tag B1)
unito = replica_a.unisci(replica_b)
print("OR-Set: A removes, B adds concurrently ->", "milk present" if unito.contiene("milk") else "milk absent",
      "(add wins)", "| the same in reversed order:", replica_b.unisci(replica_a).contiene("milk"))

# Output:
# G-Counter  commutative 2000/2000, associative 2000/2000, idempotent 2000/2000
# OR-Set     commutative 2000/2000, associative 2000/2000, idempotent 2000/2000
# PN-Counter: value 1 and 1 - same result: True
# OR-Set: A removes, B adds concurrently -> milk present (add wins) | the same in reversed order: True

The G-Counter keeps one counter per node and the merge takes the maximum of each, so no increment is lost even if communications are duplicated; the PN-Counter combines two to allow decrements. The 2,000 trials for G-Counter and OR-Set always give 2,000 out of 2,000 for commutativity, associativity and idempotence: it is not a proof, it is a check that the code keeps its promises. The PN-Counter converges to 1 whether merging in one order or another, with a repeated merge. In the OR-Set the removal deletes only the tags it has seen: replica B added "milk" with a new tag, which A did not know about, so after the merge the element is there. That is why it is called add-wins semantics, and the result is the same in any order.

The limit of CRDTs is just as concrete: the data type decides the semantics. A counter is fine for visits, a set with add-wins for a cart (with the surprise that an item removed on one device can reappear), but there is no CRDT for "the balance cannot drop below zero": global constraints require coordination.

PACELC: the trade-off CAP does not see

Abadi extends CAP with PACELC: if there is a partition (P) you choose between availability and consistency (A/C); otherwise (E, else) you choose between latency and consistency (L/C). His paper classifies Dynamo, Cassandra and Riak in their default configuration as PA/EL, that is, consistency given up in both cases; fully ACID systems like VoltDB and Megastore as PC/EC; MongoDB as PA/EC. The last classification refutes whoever writes that "PA/EC does not exist": it is in the original paper's table. The classifications hold for the versions and configurations of 2012 and must be rechecked against the ones in use.

Distributed transactions: the saga

When an operation touches several services, two-phase commit is often too costly. The saga replaces it with a sequence of local transactions, each with its own compensation: if a step fails, the compensations of the steps already completed are run. The description on microservices.io puts it this way: either all the steps succeed, or the changes of the completed steps are undone by compensating transactions.

# Saga: a sequence of local transactions, each with its compensation.
# If a step fails, the compensations of the steps already completed are run, in reverse order.
def crea_ordine(s):        s["order"] = "created"
def annulla_ordine(s):     s["order"] = "cancelled"
def riserva_scorta(s):     s["stock"] = "reserved"
def rilascia_scorta(s):    s["stock"] = "released"
def paga(s):               raise RuntimeError("payment rejected")
def rimborsa(s):           s["payment"] = "refunded"

PASSI = [(crea_ordine, annulla_ordine), (riserva_scorta, rilascia_scorta), (paga, rimborsa)]

def esegui_saga(stato: dict) -> bool:
    completati = []
    try:
        for azione, compensazione in PASSI:
            azione(stato)
            completati.append(compensazione)
        return True
    except Exception as errore:
        print("failure:", errore)
        for compensazione in reversed(completati):    # reverse order
            print("  compensating with", compensazione.__name__)
            compensazione(stato)
        return False

stato = {}
print("saga succeeded:", esegui_saga(stato))
print("final state:", stato)

# Output:
# failure: payment rejected
#   compensating with rilascia_scorta
#   compensating with annulla_ordine
# saga succeeded: False
# final state: {'order': 'cancelled', 'stock': 'released'}

The payment fails: the program compensates only the completed steps (rilascia_scorta and annulla_ordine), in reverse order, and not the payment itself, which never succeeded. The final state is consistent but not isolated: for an instant someone saw the order created and the stock reserved. A saga guarantees a late atomicity, not the isolation of a transaction; compensations must be idempotent, because they may be run again. A @Transactional annotation around a saga, as seen in some examples, is a contradiction: a local commit does not cover remote services.

What you can do now

You know that CAP forces a choice between consistency and availability only during a partition, and that the rest of the time the trade-off is between latency and consistency. You can check when a quorum guarantees intersection, recognize two concurrent changes with a vector clock, choose a CRDT when you do not want to lose updates, and compose a saga with its compensations. Above all, you know when "it depends" is the most honest answer: on the loss you can bear, the latency you can afford and the operation you are protecting.

This is the last article in the series. The starting point is the comparison of concurrency models; for the theme of ordering within a single process, see lock-free data structures, and for flows, backpressure.

I did not reach the original text of Gilbert and Lynch (the PDF was not readable from the command line): the formulation of CAP given here relies on Abadi's paper, which cites it. The simulation outputs are deterministic, except the order of the nodes in the vector clock union, which depends on Python's hash seed.