S072 · Consistency Models — Linearizable, Sequential, Eventual

#!/usr/bin/env python3
"""consistency_lab.py — a toy replicated store to feel the six models.
 
Run:  python consistency_lab.py
Deps: standard library only. Tested on 3.10+.
 
We simulate 3 replicas with async replication. A "client" library
lets you write to any replica; replication happens after a random delay.
We then check whether the observed read history is legal under
each consistency model.
"""
from __future__ import annotations
 
import random
import threading
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable
 
random.seed(42)
 
 
@dataclass
class Op:
    """A single client operation: write(k,v) or read(k) -> observed value."""
    kind: str           # "W" or "R"
    key: str
    value: str | None   # written value, OR observed value for reads
    ts_call: float      # wall time the client called
    ts_ret: float       # wall time the client saw a return
    client: str
 
 
@dataclass
class Replica:
    """One replica of a key-value store."""
    name: str
    store: dict[str, str] = field(default_factory=dict)
    # Peers to forward writes to (async).
    peers: list["Replica"] = field(default_factory=list)
    # Simulated network delay range in seconds.
    delay: tuple[float, float] = (0.01, 0.05)
 
    def write(self, key: str, value: str) -> None:
        self.store[key] = value
        # Fire-and-forget replication with a delay per peer.
        for peer in self.peers:
            d = random.uniform(*self.delay)
            threading.Timer(d, peer._apply, args=(key, value)).start()
 
    def _apply(self, key: str, value: str) -> None:
        # Last-writer-wins — the simplest (and weakest) merge policy.
        self.store[key] = value
 
    def read(self, key: str) -> str | None:
        return self.store.get(key)
 
 
def run_workload() -> list[Op]:
    """Two clients, three replicas, some writes, some reads."""
    r1, r2, r3 = Replica("r1"), Replica("r2"), Replica("r3")
    r1.peers, r2.peers, r3.peers = [r2, r3], [r1, r3], [r1, r2]
 
    history: list[Op] = []
 
    def op(client: str, kind: str, key: str,
           fn: Callable[[], str | None], value: str | None = None) -> None:
        t0 = time.time()
        result = fn()
        t1 = time.time()
        history.append(Op(kind, key, value if kind == "W" else result,
                          t0, t1, client))
 
    # t=0    C1 writes x=1 to r1
    op("C1", "W", "x", lambda: r1.write("x", "1"), value="1")
    time.sleep(0.005)
    # t=5ms  C2 reads x from r3 — probably still None (write hasn't replicated)
    op("C2", "R", "x", lambda: r3.read("x"))
    time.sleep(0.100)
    # t=105ms C2 reads x again from r3 — now sees "1"
    op("C2", "R", "x", lambda: r3.read("x"))
    # C1 writes x=2 then reads its own write from r1 — must be "2"
    op("C1", "W", "x", lambda: r1.write("x", "2"), value="2")
    op("C1", "R", "x", lambda: r1.read("x"))
    time.sleep(0.150)  # let everyone converge
    op("C2", "R", "x", lambda: r3.read("x"))
    return history
 
 
def check_linearizable(h: list[Op]) -> bool:
    """Real-time order must be respected: if op A returns before op B calls,
    then A appears before B in any valid serialisation."""
    # Simplified: check that every read sees the most recent write
    # whose ts_ret < read's ts_call.
    for op in h:
        if op.kind != "R":
            continue
        latest = None
        for w in h:
            if w.kind == "W" and w.key == op.key and w.ts_ret < op.ts_call:
                if latest is None or w.ts_ret > latest.ts_ret:
                    latest = w
        expected = latest.value if latest else None
        if op.value != expected:
            return False
    return True
 
 
def check_read_your_writes(h: list[Op]) -> bool:
    """After a client writes v to k, its own subsequent reads of k must return
    v or a later value from the same client."""
    latest_own: dict[tuple[str, str], str] = {}
    for op in h:
        if op.kind == "W":
            latest_own[(op.client, op.key)] = op.value  # type: ignore[assignment]
        else:
            expected = latest_own.get((op.client, op.key))
            if expected is not None and op.value != expected:
                return False
    return True
 
 
def check_monotonic_reads(h: list[Op]) -> bool:
    """A client's successive reads of the same key never see an older write."""
    by_client: dict[str, list[Op]] = defaultdict(list)
    for op in h:
        if op.kind == "R":
            by_client[op.client].append(op)
    # For simplicity we treat any transition value -> None as non-monotonic.
    for _c, reads in by_client.items():
        seen: set[str] = set()
        for r in reads:
            if r.value is None and seen:
                return False
            if r.value is not None:
                seen.add(r.value)
    return True
 
 
def check_eventual(h: list[Op], convergence_window: float = 0.5) -> bool:
    """After the last write, given enough time, all reads should agree."""
    last_write = max((o for o in h if o.kind == "W"), key=lambda o: o.ts_ret,
                     default=None)
    if last_write is None:
        return True
    for op in h:
        if op.kind == "R" and op.ts_call > last_write.ts_ret + convergence_window:
            if op.value != last_write.value:
                return False
    return True
 
 
def main() -> None:
    h = run_workload()
    print(f"\n{'time (ms)':>10}  {'client':>6}  op                    result")
    t0 = h[0].ts_call
    for op in h:
        rel = int((op.ts_call - t0) * 1000)
        detail = (f"W({op.key}={op.value})" if op.kind == "W"
                  else f"R({op.key})->{op.value!r}")
        print(f"{rel:>10}  {op.client:>6}  {detail:<20}")
    print()
    print("Legal under linearizable?     ", check_linearizable(h))
    print("Legal under read-your-writes? ", check_read_your_writes(h))
    print("Legal under monotonic reads?  ", check_monotonic_reads(h))
    print("Legal under eventual?         ", check_eventual(h))
 
 
if __name__ == "__main__":
    main()