diff --git a/wizard-work/null/crdtkit/README-crdtkit.md b/wizard-work/null/crdtkit/README-crdtkit.md new file mode 100644 index 0000000..8af4b21 --- /dev/null +++ b/wizard-work/null/crdtkit/README-crdtkit.md @@ -0,0 +1,84 @@ +# crdtkit — state-based CRDTs with proven merge laws + +Three dependency-free CvRDTs in `orcrdt.py`, and a test suite that checks the merge +laws two ways: by random gossip simulation, and by **mutation** — deliberately +broken merges must be *caught* by the same check, or the check is decoration. + + crdtkit/orcrdt.py ORSet, PNCounter, LWWRegister, CRDTError (stdlib only) + crdtkit/test_crdt.py 29 unittest cases: laws, semantics, serialization, + convergence, mutation teeth, differential oracle + crdtkit/evidence_run.py prints the evidence matrix; exit 0 iff every claim holds + crdtkit/evidence.txt captured output of the run below + +Re-run (any cwd): + + python3 /root/troupe-standup/wizard-work/null/crdtkit/test_crdt.py + python3 /root/troupe-standup/wizard-work/null/crdtkit/evidence_run.py + +## Semantics + +| type | state | merge | guarantee | +|---|---|---|---| +| `ORSet` | `elem -> {tag}`, tombstone set of tags; tag = `(replica, counter)` | union of both sets | add-wins: a `remove` kills only the tags it observed | +| `PNCounter` | per-replica `inc`/`dec` vectors | pointwise **max** | each replica writes only its own slot, so max is a join | +| `LWWRegister` | `(value, stamp)`; stamp = `(timestamp:int, replica:str)` | max stamp | winner independent of merge order; equal stamp + differing value is refused | + +`merge` is pure — it never mutates `self` or its arguments — and every state has a +canonical `state_key()` plus a `to_state()/dumps()/loads()` JSON round-trip. + +## Measured evidence (verbatim `evidence.txt`) + + orcrdt evidence run + python 3.14.7 on Linux + seeds 0..15, 3 replicas x 80 ops x 4 gossip rounds (5x150 for the big one) + + [A] convergence: every replica equals the join of all pre-gossip states + ORSet 16/16 seeds 5-replica ensemble: converged + PNCounter 16/16 seeds 5-replica ensemble: converged + LWWRegister 16/16 seeds 5-replica ensemble: converged + + [B] mutation teeth: a broken merge must be caught by the same check + ORSetDropsForeignTombstones caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + PNCounterDropsDecrements caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + LWWLastArgumentWins caught at 10/16 seeds; semantic divergence: yes; real class green on those seeds: yes + ORSetNoOpMerge caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + PNCounterSumMerge caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + + [C] differential vs independent reference model (OR-Set) + 16/16 seeds agree (both implementations, all replicas) + + FAILURES: none + +Additional runs (measured, not claimed): suite exits 0 under `PYTHONHASHSEED` in +{0, 1, 7, 42, 12345} (no dict-iteration-order dependence) and under `python3 -O`. + +Negative control (measured): a copy of this kit whose `ORSet.merge` drops foreign +tombstones — a one-line change in the merge — exits **1** with 7 failures, while the +unmodified kit exits 0. The CHECK is therefore a real test, not a tautology. + +## What each block proves, and what it does not + +* `[A]` — replicas that gossip **serialized snapshots** in random order, with + duplicate deliveries, all end at the exact join of their pre-gossip states, in + both state and semantic value. 3 replicas x 80 ops x 16 seeds, plus a 5-replica + x 150-op ensemble. Not proven: any real transport, reordering under partitions, + or Byzantine behaviour. +* `[B]` — five wrong merges (drop foreign tombstones, drop decrements, last-argument + wins, ignore the argument, sum instead of max) all go red on the same check, while + the real class stays green on exactly those seeds. This is the part that makes + `[A]` evidence rather than a green light that would also shine on broken code. + `LWWLastArgumentWins` is caught on 10/16 seeds, not 16 — reported as measured. +* `[C]` — `ORSet` agrees with `RefORSet`, a separately written naive model with a + different state layout (flat `(elem, replica, counter)` tag set). Agreement is + cross-implementation, not self-confirmation. Not proven: agreement on workloads + outside the harness (only 5 distinct elements, 3 replicas, one op mix). + +## Honest limits + +* Elements must be JSON scalars (`str/int/float/bool/None`) — deliberate, so + `to_state()` is lossless rather than `repr`-based. +* `LWWRegister` assumes replica ids are unique per node; two nodes sharing an id + and writing at the same timestamp raise `CRDTError` instead of silently diverging. +* Tombstones and per-replica vector slots are never garbage-collected: state grows + with history. No delta-CRDT or anti-entropy protocol is provided. +* Merge is whole-state (`O(state)`), not incremental. diff --git a/wizard-work/null/crdtkit/evidence.txt b/wizard-work/null/crdtkit/evidence.txt new file mode 100644 index 0000000..50e99e6 --- /dev/null +++ b/wizard-work/null/crdtkit/evidence.txt @@ -0,0 +1,20 @@ +orcrdt evidence run +python 3.14.7 on Linux +seeds 0..15, 3 replicas x 80 ops x 4 gossip rounds (5x150 for the big one) + +[A] convergence: every replica equals the join of all pre-gossip states + ORSet 16/16 seeds 5-replica ensemble: converged + PNCounter 16/16 seeds 5-replica ensemble: converged + LWWRegister 16/16 seeds 5-replica ensemble: converged + +[B] mutation teeth: a broken merge must be caught by the same check + ORSetDropsForeignTombstones caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + PNCounterDropsDecrements caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + LWWLastArgumentWins caught at 10/16 seeds; semantic divergence: yes; real class green on those seeds: yes + ORSetNoOpMerge caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + PNCounterSumMerge caught at 16/16 seeds; semantic divergence: yes; real class green on those seeds: yes + +[C] differential vs independent reference model (OR-Set) + 16/16 seeds agree (both implementations, all replicas) + +FAILURES: none diff --git a/wizard-work/null/crdtkit/evidence_run.py b/wizard-work/null/crdtkit/evidence_run.py new file mode 100644 index 0000000..0c6911d --- /dev/null +++ b/wizard-work/null/crdtkit/evidence_run.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""evidence_run.py -- reproducible evidence for orcrdt; exit 0 iff all claims hold. + +Prints, in one pass: + [A] random-gossip convergence over every seed, per CRDT family + [B] mutation teeth: each deliberately broken merge must be caught, and the real + class must pass exactly the seeds where its mutant failed + [C] differential agreement with an independently written reference model + +Nothing here trusts the library: every line is recomputed from the simulation +harness in test_crdt.py, and the script's own exit code is the verdict. +""" + +from __future__ import annotations + +import os +import platform +import random +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import test_crdt as T # noqa: E402 +from orcrdt import LWWRegister, ORSet, PNCounter # noqa: E402 + +FAMILIES = ((ORSet, T.ORSetDropsForeignTombstones), + (PNCounter, T.PNCounterDropsDecrements), + (LWWRegister, T.LWWLastArgumentWins), + (ORSet, T.ORSetNoOpMerge), + (PNCounter, T.PNCounterSumMerge)) + + +def main(): + failures = [] + print("orcrdt evidence run") + print("python %s on %s" % (platform.python_version(), platform.system())) + print("seeds 0..%d, 3 replicas x 80 ops x 4 gossip rounds (5x150 for the big one)" + % (max(T.SEEDS),)) + print() + + print("[A] convergence: every replica equals the join of all pre-gossip states") + for real in dict.fromkeys(real for real, _ in FAMILIES): + passed = sum(1 for s in T.SEEDS if T.check_convergence(real, s)) + big = all(T.check_convergence(real, s, n=5, ops=150, rounds=6) + for s in (0, 1, 2)) + print(" %-12s %2d/%d seeds 5-replica ensemble: %s" + % (real.__name__, passed, len(list(T.SEEDS)), + "converged" if big else "DIVERGED")) + if passed != len(list(T.SEEDS)) or not big: + failures.append("convergence %s" % real.__name__) + print() + + print("[B] mutation teeth: a broken merge must be caught by the same check") + for real, mutant in FAMILIES: + hits = T.mutant_detected(mutant) + real_ok = all(T.check_convergence(real, s) for s in hits) + value_wrong = any(not T.check_convergence_detail(mutant, s)["value_ok"] + for s in hits) + print(" %-32s caught at %2d/%d seeds; semantic divergence: %-3s; " + "real class green on those seeds: %s" + % (mutant.__name__, len(hits), len(list(T.SEEDS)), + "yes" if value_wrong else "no", "yes" if real_ok else "NO")) + if not hits or not real_ok: + failures.append("teeth %s" % mutant.__name__) + print() + + print("[C] differential vs independent reference model (OR-Set)") + agree = 0 + for seed in range(16): + rng = random.Random(seed * 31 + 5) + live = [ORSet("r%d" % i) for i in range(3)] + ref = [T.RefORSet("r%d" % i) for i in range(3)] + for _ in range(120): + i, elem = rng.randrange(3), rng.choice(T.ELEMS) + if rng.random() < 0.55: + live[i].add(elem) + ref[i].add(elem) + else: + live[i].remove(elem) + ref[i].remove(elem) + for dst, src in [(rng.randrange(3), rng.randrange(3)) for _ in range(60)]: + if dst == src: + continue + live[dst] = live[dst].merge(live[src]) + ref[dst] = ref[dst].merge(ref[src]) + if live[0].elements() == ref[0].elements() == live[1].elements() == live[2].elements(): + agree += 1 + print(" %d/16 seeds agree (both implementations, all replicas)" % agree) + if agree != 16: + failures.append("differential") + + print() + print("FAILURES: %s" % (", ".join(failures) if failures else "none")) + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/wizard-work/null/crdtkit/orcrdt.py b/wizard-work/null/crdtkit/orcrdt.py new file mode 100644 index 0000000..40af1f9 --- /dev/null +++ b/wizard-work/null/crdtkit/orcrdt.py @@ -0,0 +1,412 @@ +#!/usr/bin/env python3 +"""orcrdt -- dependency-free state-based CRDT kit (OR-Set, PN-Counter, LWW-Register). + +Design rules +------------ +* State-based (CvRDT): every replica exposes ``merge(other, ...) -> new`` implementing + a join over a semilattice. ``merge`` is PURE -- it never mutates ``self`` or its + arguments, so a replica can re-deliver a snapshot any number of times. +* A merge that is not commutative / associative / idempotent makes replicas diverge. + That is not a matter of opinion, so ``test_crdt.py`` checks it two ways: random + gossip simulation over many seeds, and mutation -- deliberately broken merges must + be *detected* by the same check. +* Elements of ``ORSet`` must be JSON scalars (str/int/float/bool/None). This is a + deliberate constraint: it makes ``to_state()`` lossless instead of repr-based. +* No third-party imports. Python 3.9+. + +Public surface +-------------- +ORSet add-wins observed-remove set +PNCounter increment/decrement counter (per-replica vectors, pointwise max) +LWWRegister last-writer-wins register, stamp = (timestamp, replica) +CRDTError raised on malformed state or a broken local operation +""" + +from __future__ import annotations + +import json + +__all__ = ["ORSet", "PNCounter", "LWWRegister", "CRDTError"] + +_SCALARS = (str, int, float, bool, type(None)) + + +class CRDTError(ValueError): + """Malformed state, non-scalar element, or a non-monotone local operation.""" + + +def _check_scalar(elem, where): + if isinstance(elem, _SCALARS): + return elem + raise CRDTError( + "%s: element must be a JSON scalar (str/int/float/bool/None), got %r" + % (where, type(elem).__name__) + ) + + +def _canon(elem): + """Canonical, order-stable, comparable form of a scalar element.""" + return json.dumps(elem, sort_keys=True) + + +def _sorted_tags(tags): + return sorted(tags, key=lambda t: (t[0], t[1])) + + +def _pmax(a, b): + """Pointwise max over the union of two non-negative-int vectors.""" + out = dict(a) + for k, v in b.items(): + if v > out.get(k, 0): + out[k] = v + return out + + +# --------------------------------------------------------------------------- +# OR-Set +# --------------------------------------------------------------------------- +class ORSet: + """Observed-Remove Set with add-wins semantics. + + State = (``_adds``: elem -> set of unique tags, ``_removed``: set of tags). + A tag is ``(replica, counter)`` and is unique per local ``add``. + + ``contains(e)`` is true iff ``e`` has at least one tag that no ``remove`` has + tombstoned. ``remove(e)`` tombstones only the tags the caller had *observed*, + so a concurrent ``add`` survives the merge -- that is the add-wins rule. + """ + + __slots__ = ("replica", "_adds", "_removed", "_clock") + + def __init__(self, replica="r0"): + if not isinstance(replica, str) or not replica: + raise CRDTError("replica id must be a non-empty string") + self.replica = replica + self._adds = {} # elem -> set(tag) + self._removed = set() # set(tag) + self._clock = 0 + + # -- local mutations ---------------------------------------------------- + def add(self, elem): + _check_scalar(elem, "ORSet.add") + self._clock += 1 + tag = (self.replica, self._clock) + self._adds.setdefault(elem, set()).add(tag) + return tag + + def remove(self, elem): + """Tombstone every tag of ``elem`` observed locally. Returns tags removed.""" + _check_scalar(elem, "ORSet.remove") + tags = self._adds.get(elem) + if not tags: + return 0 + fresh = [t for t in tags if t not in self._removed] + self._removed.update(fresh) + return len(fresh) + + # -- queries ------------------------------------------------------------ + def contains(self, elem): + return any(t not in self._removed for t in self._adds.get(elem, ())) + + def elements(self): + return frozenset(e for e in self._adds if self.contains(e)) + + def value_key(self): + return tuple(sorted(_canon(e) for e in self.elements())) + + def __contains__(self, elem): + return self.contains(elem) + + def __len__(self): + return len(self.elements()) + + def __iter__(self): + return iter(sorted(self.elements(), key=_canon)) + + # -- merge / state ------------------------------------------------------ + def merge(self, *others): + out = ORSet(self.replica) + states = (self,) + others + for st in states: + if not isinstance(st, ORSet): + raise CRDTError("ORSet.merge: expected ORSet, got %r" % type(st).__name__) + for e, tags in st._adds.items(): + out._adds.setdefault(e, set()).update(tags) + out._removed.update(st._removed) + out._clock = max(st._clock for st in states) + return out + + def state_key(self): + adds = tuple( + sorted( + (_canon(e), tuple(_sorted_tags(t))) for e, t in self._adds.items() + ) + ) + return (adds, tuple(_sorted_tags(self._removed))) + + def __eq__(self, other): + return isinstance(other, ORSet) and self.state_key() == other.state_key() + + def __hash__(self): + return hash(self.state_key()) + + def to_state(self): + return { + "type": "ORSet", + "replica": self.replica, + "clock": self._clock, + "adds": [ + [_canon(e), [[t[0], t[1]] for t in _sorted_tags(tags)]] + for e, tags in sorted(self._adds.items(), key=lambda kv: _canon(kv[0])) + ], + "removed": [[t[0], t[1]] for t in _sorted_tags(self._removed)], + } + + @classmethod + def from_state(cls, d): + try: + if d["type"] != "ORSet": + raise CRDTError("ORSet.from_state: wrong type tag %r" % (d["type"],)) + out = cls(d["replica"]) + out._clock = int(d["clock"]) + for raw_e, raw_tags in d["adds"]: + elem = json.loads(raw_e) + _check_scalar(elem, "ORSet.from_state") + out._adds.setdefault(elem, set()).update( + (t[0], int(t[1])) for t in raw_tags + ) + out._removed = {(t[0], int(t[1])) for t in d["removed"]} + if out._clock < 0 or any(c <= 0 for _, c in out._removed): + raise CRDTError("ORSet.from_state: non-positive clock/counter") + return out + except (KeyError, TypeError, ValueError) as exc: + if isinstance(exc, CRDTError): + raise + raise CRDTError("ORSet.from_state: malformed state: %s" % exc) from exc + + def dumps(self): + return json.dumps(self.to_state(), sort_keys=True) + + @classmethod + def loads(cls, s): + try: + d = json.loads(s) + except ValueError as exc: + raise CRDTError("ORSet.loads: not JSON: %s" % exc) from exc + if not isinstance(d, dict): + raise CRDTError("ORSet.loads: state must be a JSON object") + return cls.from_state(d) + + +# --------------------------------------------------------------------------- +# PN-Counter +# --------------------------------------------------------------------------- +class PNCounter: + """Increment/decrement counter. + + State = two per-replica vectors (``inc``, ``dec``); merge = pointwise max. + Each replica writes only its own slot, which is what makes pointwise max a + correct join (a replica's own slot only ever grows). + """ + + __slots__ = ("replica", "inc", "dec") + + def __init__(self, replica="r0"): + if not isinstance(replica, str) or not replica: + raise CRDTError("replica id must be a non-empty string") + self.replica = replica + self.inc = {} + self.dec = {} + + def _bump(self, vec, n): + if not isinstance(n, int) or isinstance(n, bool) or n < 0: + raise CRDTError("PNCounter: delta must be a non-negative int, got %r" % (n,)) + vec[self.replica] = vec.get(self.replica, 0) + n + + def increment(self, n=1): + self._bump(self.inc, n) + + def decrement(self, n=1): + self._bump(self.dec, n) + + def value(self): + return sum(self.inc.values()) - sum(self.dec.values()) + + def value_key(self): + return self.value() + + def merge(self, *others): + out = PNCounter(self.replica) + for st in (self,) + others: + if not isinstance(st, PNCounter): + raise CRDTError("PNCounter.merge: expected PNCounter, got %r" % type(st).__name__) + out.inc = _pmax(out.inc, st.inc) + out.dec = _pmax(out.dec, st.dec) + return out + + def state_key(self): + return (tuple(sorted(self.inc.items())), tuple(sorted(self.dec.items()))) + + def __eq__(self, other): + return isinstance(other, PNCounter) and self.state_key() == other.state_key() + + def __hash__(self): + return hash(self.state_key()) + + def to_state(self): + return { + "type": "PNCounter", + "replica": self.replica, + "inc": sorted(self.inc.items()), + "dec": sorted(self.dec.items()), + } + + @classmethod + def from_state(cls, d): + try: + if d["type"] != "PNCounter": + raise CRDTError("PNCounter.from_state: wrong type tag %r" % (d["type"],)) + out = cls(d["replica"]) + for key in ("inc", "dec"): + vec = {} + for k, v in d[key]: + if not isinstance(k, str) or not isinstance(v, int) or v < 0: + raise CRDTError("PNCounter.from_state: bad entry in %s" % key) + vec[k] = v + setattr(out, key, vec) + return out + except (KeyError, TypeError, ValueError) as exc: + if isinstance(exc, CRDTError): + raise + raise CRDTError("PNCounter.from_state: malformed state: %s" % exc) from exc + + def dumps(self): + return json.dumps(self.to_state(), sort_keys=True) + + @classmethod + def loads(cls, s): + try: + d = json.loads(s) + except ValueError as exc: + raise CRDTError("PNCounter.loads: not JSON: %s" % exc) from exc + if not isinstance(d, dict): + raise CRDTError("PNCounter.loads: state must be a JSON object") + return cls.from_state(d) + + +# --------------------------------------------------------------------------- +# LWW-Register +# --------------------------------------------------------------------------- +class LWWRegister: + """Last-writer-wins register. + + Stamp = ``(timestamp: int, replica: str)``. Merge takes the maximum stamp, so + the winner is total-order-determined and independent of argument order. Local + writes must advance the stamp strictly (``set`` raises otherwise), and two + different values at an *identical* stamp are refused rather than silently + broken -- that only happens if two nodes reuse one replica id. + """ + + __slots__ = ("replica", "_value", "_stamp") + + def __init__(self, replica="r0"): + if not isinstance(replica, str) or not replica: + raise CRDTError("replica id must be a non-empty string") + self.replica = replica + self._value = None + self._stamp = (-1, "") + + def set(self, value, timestamp): + if not isinstance(timestamp, int) or isinstance(timestamp, bool): + raise CRDTError("LWWRegister.set: timestamp must be an int") + stamp = (timestamp, self.replica) + if stamp <= self._stamp: + raise CRDTError( + "LWWRegister.set: non-monotone stamp %r <= %r" % (stamp, self._stamp) + ) + self._value = value + self._stamp = stamp + return stamp + + def get(self): + return self._value + + @property + def stamp(self): + return self._stamp + + def value_key(self): + return self._value + + def merge(self, *others): + out = LWWRegister(self.replica) + out._value, out._stamp = self._value, self._stamp + for st in (self,) + others: + if not isinstance(st, LWWRegister): + raise CRDTError( + "LWWRegister.merge: expected LWWRegister, got %r" % type(st).__name__ + ) + if st._stamp > out._stamp: + out._value, out._stamp = st._value, st._stamp + elif st._stamp == out._stamp and st._value != out._value: + raise CRDTError( + "LWWRegister.merge: conflicting values at identical stamp %r" + % (st._stamp,) + ) + return out + + def state_key(self): + return (self._stamp, _canon(self._value) if isinstance(self._value, _SCALARS) else repr(self._value)) + + def __eq__(self, other): + return isinstance(other, LWWRegister) and self.state_key() == other.state_key() + + def __hash__(self): + return hash(self.state_key()) + + def to_state(self): + return { + "type": "LWWRegister", + "replica": self.replica, + "stamp": [self._stamp[0], self._stamp[1]], + "value": self._value, + } + + @classmethod + def from_state(cls, d): + try: + if d["type"] != "LWWRegister": + raise CRDTError("LWWRegister.from_state: wrong type tag %r" % (d["type"],)) + out = cls(d["replica"]) + out._stamp = (int(d["stamp"][0]), d["stamp"][1]) + out._value = d["value"] + return out + except (KeyError, TypeError, ValueError, IndexError) as exc: + if isinstance(exc, CRDTError): + raise + raise CRDTError("LWWRegister.from_state: malformed state: %s" % exc) from exc + + def dumps(self): + return json.dumps(self.to_state(), sort_keys=True) + + @classmethod + def loads(cls, s): + try: + d = json.loads(s) + except ValueError as exc: + raise CRDTError("LWWRegister.loads: not JSON: %s" % exc) from exc + if not isinstance(d, dict): + raise CRDTError("LWWRegister.loads: state must be a JSON object") + return cls.from_state(d) + + +if __name__ == "__main__": + # 12-line smoke illustration: concurrent remove loses to concurrent add. + a, b = ORSet("a"), ORSet("b") + a.add("x") + b = ORSet.loads(a.dumps()) + a.remove("x") + b.add("x") + m = a.merge(b) + print("orcrdt smoke: a.elements=%r b.elements=%r merged=%r" + % (sorted(a), sorted(b), sorted(m))) diff --git a/wizard-work/null/crdtkit/test_crdt.py b/wizard-work/null/crdtkit/test_crdt.py new file mode 100644 index 0000000..e7d7ee5 --- /dev/null +++ b/wizard-work/null/crdtkit/test_crdt.py @@ -0,0 +1,506 @@ +#!/usr/bin/env python3 +"""test_crdt.py -- property + mutation tests for orcrdt. + +Run: python3 test_crdt.py (exit 0 iff every test passes) + python3 test_crdt.py -v + +Two kinds of evidence: +1. SIMULATION -- N replicas take random local ops, then gossip their serialized + snapshots in random order with duplicate deliveries. Every replica must end at + the exact join of all pre-gossip states (state AND semantic value). +2. MUTATION TEETH -- the same convergence check is pointed at deliberately broken + merge implementations; if it does not go red for them, the check is worthless + and this suite fails. +""" + +from __future__ import annotations + +import os +import random +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from orcrdt import CRDTError, LWWRegister, ORSet, PNCounter # noqa: E402 + +ELEMS = ["alpha", "beta", "gamma", "delta", "epsilon"] +SEEDS = range(16) + + +# --------------------------------------------------------------------------- +# simulation harness +# --------------------------------------------------------------------------- +def _family(kind): + """Map a CRDT class (or a mutant subclass of one) to its base family.""" + for base in (ORSet, PNCounter, LWWRegister): + if issubclass(kind, base): + return base + raise TypeError("unknown CRDT family: %r" % (kind,)) + + +def _local_op(kind, rng, rep): + fam = _family(kind) + if fam is ORSet: + elem = rng.choice(ELEMS) + if rng.random() < 0.55: + rep.add(elem) + else: + rep.remove(elem) + elif fam is PNCounter: + d = rng.choice([1, 1, 1, 2, 3, -1, -2]) + if d > 0: + rep.increment(d) + else: + rep.decrement(-d) + else: + ts = rep.stamp[0] + 1 + rng.randrange(3) + rep.set(rng.randrange(50), ts) + + +def make_replicas(kind, n, ops, seed): + rng = random.Random(seed) + reps = [kind("r%d" % i) for i in range(n)] + for _ in range(ops): + _local_op(kind, rng, rng.choice(reps)) + return reps + + +def gossip(kind, reps, seed, rounds=4): + """Random pairwise merges with duplicate deliveries, over serialized copies.""" + rng = random.Random(seed * 7919 + 13) + n = len(reps) + live = [kind.loads(r.dumps()) for r in reps] + for _ in range(rounds * n * 3): + dst, src = rng.randrange(n), rng.randrange(n) + if dst == src: + continue + live[dst] = live[dst].merge(live[src]) + return live + + +def check_convergence(kind, seed, n=3, ops=80, rounds=4): + """True iff gossip converges to the join of the pre-gossip replica states.""" + reps = make_replicas(kind, n, ops, seed) + snaps = [kind.loads(r.dumps()) for r in reps] + + full = snaps[0] + for s in snaps[1:]: + full = full.merge(s) + + live = gossip(kind, reps, seed, rounds) + state_ok = all(x.state_key() == full.state_key() for x in live) + value_ok = all(x.value_key() == full.value_key() for x in live) + join_ok = full.state_key() == full.merge(*snaps).state_key() + return state_ok and value_ok and join_ok + + +def check_convergence_detail(kind, seed, **kw): + reps = make_replicas(kind, kw.get("n", 3), kw.get("ops", 80), seed) + snaps = [kind.loads(r.dumps()) for r in reps] + full = snaps[0] + for s in snaps[1:]: + full = full.merge(s) + live = gossip(kind, reps, seed, kw.get("rounds", 4)) + return { + "state_ok": all(x.state_key() == full.state_key() for x in live), + "value_ok": all(x.value_key() == full.value_key() for x in live), + } + + +# --------------------------------------------------------------------------- +# deliberately broken merges (mutation testing of the harness above) +# --------------------------------------------------------------------------- +class ORSetDropsForeignTombstones(ORSet): + """BUG: merge keeps only self's tombstone set -> removed elements resurrect.""" + + def merge(self, *others): + out = super().merge(*others) + out._removed = set(self._removed) + return out + + +class PNCounterDropsDecrements(PNCounter): + """BUG: merge folds inc by max but drops the others' dec vectors.""" + + def merge(self, *others): + out = PNCounter(self.replica) + out.inc, out.dec = dict(self.inc), dict(self.dec) + for st in others: + for k, v in st.inc.items(): + if v > out.inc.get(k, 0): + out.inc[k] = v + return out + + +class LWWLastArgumentWins(LWWRegister): + """BUG: merge always takes the last argument -> order dependent.""" + + def merge(self, *others): + out = LWWRegister(self.replica) + out._value, out._stamp = self._value, self._stamp + for st in others: + out._value, out._stamp = st._value, st._stamp + return out + + +class ORSetNoOpMerge(ORSet): + """BUG: merge ignores its argument (no propagation) -> not commutative.""" + + def merge(self, *others): + out = ORSet(self.replica) + out._adds = {e: set(t) for e, t in self._adds.items()} + out._removed = set(self._removed) + out._clock = self._clock + return out + + +class PNCounterSumMerge(PNCounter): + """BUG: merge adds the vectors instead of taking pointwise max. + + Commutative and associative, so naive gossip tests can miss it -- it fails + IDEMPOTENCE, and any duplicate delivery then inflates the value. + """ + + def merge(self, *others): + out = PNCounter(self.replica) + out.inc, out.dec = dict(self.inc), dict(self.dec) + for st in others: + for k, v in st.inc.items(): + out.inc[k] = out.inc.get(k, 0) + v + for k, v in st.dec.items(): + out.dec[k] = out.dec.get(k, 0) + v + return out + + +def mutant_detected(mutant_cls, seeds=SEEDS): + """True iff the convergence check goes red (or raises) for >=1 seed.""" + hits = [] + for seed in seeds: + try: + ok = check_convergence(mutant_cls, seed) + except CRDTError: + ok = False + if not ok: + hits.append(seed) + return hits + + +class RefORSet: + """Independent naive add-wins OR-Set, written from the paper semantics. + + Different layout on purpose: a flat set of ``(elem, replica, counter)`` tags + plus a tombstone set, so agreeing with ``orcrdt.ORSet`` is evidence and not + self-confirmation. Test-only; not exported by the library. + """ + + def __init__(self, replica): + self.replica = replica + self.adds = set() + self.removed = set() + self.clock = 0 + + def add(self, elem): + self.clock += 1 + self.adds.add((elem, self.replica, self.clock)) + + def remove(self, elem): + for tag in [t for t in self.adds if t[0] == elem]: + self.removed.add(tag) + + def elements(self): + return frozenset(t[0] for t in self.adds if t not in self.removed) + + def merge(self, *others): + out = RefORSet(self.replica) + for o in (self,) + others: + out.adds |= o.adds + out.removed |= o.removed + out.clock = max(out.clock, o.clock) + return out + + +# --------------------------------------------------------------------------- +# tests +# --------------------------------------------------------------------------- +class TestMergeLaws(unittest.TestCase): + def _sample(self, kind, seed): + reps = make_replicas(kind, 3, 40, seed) + return [kind.loads(r.dumps()) for r in reps] + + def test_commutative_and_associative(self): + for kind in (ORSet, PNCounter, LWWRegister): + for seed in range(8): + a, b, c = self._sample(kind, seed) + self.assertEqual(a.merge(b).state_key(), b.merge(a).state_key(), + "%s not commutative (seed %d)" % (kind.__name__, seed)) + self.assertEqual(a.merge(b).merge(c).state_key(), + a.merge(b.merge(c)).state_key(), + "%s not associative (seed %d)" % (kind.__name__, seed)) + + def test_idempotent(self): + for kind in (ORSet, PNCounter, LWWRegister): + for seed in range(8): + for x in self._sample(kind, seed): + self.assertEqual(x.merge(x).state_key(), x.state_key(), + "%s not idempotent (seed %d)" % (kind.__name__, seed)) + self.assertEqual(x.merge(x, x, x).state_key(), x.state_key()) + + def test_merge_is_pure(self): + for kind in (ORSet, PNCounter, LWWRegister): + a, b, _ = self._sample(kind, 3) + before = (a.state_key(), b.state_key()) + a.merge(b) + self.assertEqual((a.state_key(), b.state_key()), before, + "%s.merge mutated an input" % kind.__name__) + + def test_duplicate_delivery_harmless(self): + for kind in (ORSet, PNCounter, LWWRegister): + a, b, _ = self._sample(kind, 5) + once = a.merge(b) + many = a + for _ in range(5): + many = many.merge(b) + self.assertEqual(many.state_key(), once.state_key()) + + def test_type_confusion_refused(self): + self.assertRaises(CRDTError, ORSet("a").merge, PNCounter("b")) + self.assertRaises(CRDTError, PNCounter("a").merge, LWWRegister("b")) + self.assertRaises(CRDTError, LWWRegister("a").merge, ORSet("b")) + + +class TestORSetSemantics(unittest.TestCase): + def test_add_wins_on_concurrent_add_remove(self): + a = ORSet("a") + a.add("x") + b = ORSet.loads(a.dumps()) + a.remove("x") + b.add("x") + self.assertFalse(a.contains("x")) + self.assertTrue(b.contains("x")) + self.assertTrue(a.merge(b).contains("x"), "add-wins violated") + self.assertTrue(b.merge(a).contains("x")) + + def test_remove_only_kills_observed_tags(self): + a = ORSet("a") + a.add("x") + b = ORSet.loads(a.dumps()) + b.remove("x") + a.add("x") # new tag, not observed by b + self.assertTrue(a.merge(b).contains("x")) + self.assertFalse(b.merge(b).contains("x")) + + def test_remove_then_merge_then_still_absent(self): + a = ORSet("a") + a.add("x") + b = a.merge(ORSet("b")) + b.remove("x") + self.assertFalse(a.merge(b).contains("x")) + + def test_elements_and_iteration(self): + a = ORSet("a") + for e in ("delta", "alpha", "gamma"): + a.add(e) + self.assertEqual(sorted(a), ["alpha", "delta", "gamma"]) + self.assertEqual(len(a), 3) + self.assertIn("alpha", a) + a.remove("alpha") + self.assertEqual(sorted(a), ["delta", "gamma"]) + + def test_rejects_non_scalar_elements(self): + a = ORSet("a") + self.assertRaises(CRDTError, a.add, ["list"]) + self.assertRaises(CRDTError, a.add, {"k": 1}) + self.assertRaises(CRDTError, ORSet, "") + self.assertRaises(CRDTError, a.remove, ["list"]) + + +class TestPNCounter(unittest.TestCase): + def test_value_and_merge(self): + a, b = PNCounter("a"), PNCounter("b") + a.increment(5) + a.decrement(2) + b.increment(3) + self.assertEqual(a.value(), 3) + self.assertEqual(b.value(), 3) + m = a.merge(b) + self.assertEqual(m.value(), 6) + self.assertEqual(m.state_key(), b.merge(a).state_key()) + + def test_pointwise_max_not_sum(self): + a = PNCounter("a") + a.increment(4) + again = PNCounter.loads(a.dumps()) + self.assertEqual(a.merge(again).value(), 4, "merge double-counted a slot") + + def test_rejects_negative_delta(self): + a = PNCounter("a") + self.assertRaises(CRDTError, a.increment, -1) + self.assertRaises(CRDTError, a.decrement, -3) + + +class TestLWWRegister(unittest.TestCase): + def test_max_stamp_wins_regardless_of_order(self): + a, b = LWWRegister("a"), LWWRegister("b") + a.set("old", 1) + b.set("new", 2) + self.assertEqual(a.merge(b).get(), "new") + self.assertEqual(b.merge(a).get(), "new") + + def test_replica_tiebreak(self): + a, b = LWWRegister("a"), LWWRegister("b") + a.set("from-a", 7) + b.set("from-b", 7) + self.assertEqual(a.merge(b).get(), "from-b") # ("b" > "a") at equal ts + self.assertEqual(b.merge(a).get(), "from-b") + + def test_non_monotone_write_refused(self): + a = LWWRegister("a") + a.set("v1", 5) + self.assertRaises(CRDTError, a.set, "v2", 5) + self.assertRaises(CRDTError, a.set, "v2", 4) + + def test_identical_stamp_conflict_refused(self): + a = LWWRegister("a") + a.set("v1", 3) + impostor = LWWRegister.loads(a.dumps()) + impostor._value = "v2" + self.assertRaises(CRDTError, a.merge, impostor) + + +class TestSerialization(unittest.TestCase): + def test_roundtrip_preserves_state(self): + for kind in (ORSet, PNCounter, LWWRegister): + for seed in range(6): + for r in make_replicas(kind, 3, 30, seed): + self.assertEqual(kind.loads(r.dumps()).state_key(), r.state_key(), + "%s roundtrip lost state" % kind.__name__) + + def test_merge_across_serialization_boundary(self): + for kind in (ORSet, PNCounter, LWWRegister): + reps = make_replicas(kind, 3, 50, 11) + direct = reps[0].merge(reps[1]) + over_wire = kind.loads(reps[0].dumps()).merge(kind.loads(reps[1].dumps())) + self.assertEqual(over_wire.state_key(), direct.state_key()) + self.assertEqual(over_wire.value_key(), direct.value_key()) + + def test_malformed_state_refused(self): + for kind in (ORSet, PNCounter, LWWRegister): + self.assertRaises(CRDTError, kind.loads, "not json") + self.assertRaises(CRDTError, kind.loads, "[1, 2, 3]") + self.assertRaises(CRDTError, kind.loads, '{"type": "Other"}') + + +class TestDifferentialReference(unittest.TestCase): + """Compare OR-Set against an independently written naive reference model.""" + + def test_orset_matches_independent_reference(self): + for seed in range(16): + rng = random.Random(seed * 31 + 5) + live = [ORSet("r%d" % i) for i in range(3)] + ref = [RefORSet("r%d" % i) for i in range(3)] + for _ in range(120): + i, elem = rng.randrange(3), rng.choice(ELEMS) + if rng.random() < 0.55: + live[i].add(elem) + ref[i].add(elem) + else: + live[i].remove(elem) + ref[i].remove(elem) + schedule = [(rng.randrange(3), rng.randrange(3)) for _ in range(60)] + for dst, src in schedule: + if dst == src: + continue + live[dst] = live[dst].merge(live[src]) + ref[dst] = ref[dst].merge(ref[src]) + self.assertEqual(live[0].elements(), ref[0].elements(), + "differential mismatch at seed %d" % seed) + for x in live[1:]: + self.assertEqual(x.elements(), live[0].elements(), + "replica split at seed %d" % seed) + + +class TestConvergence(unittest.TestCase): + def test_random_gossip_converges_all_kinds(self): + for kind in (ORSet, PNCounter, LWWRegister): + for seed in SEEDS: + detail = check_convergence_detail(kind, seed) + self.assertTrue( + detail["state_ok"] and detail["value_ok"], + "%s diverged at seed %d: %r" % (kind.__name__, seed, detail), + ) + self.assertTrue(check_convergence(kind, seed), + "%s failed join check at seed %d" % (kind.__name__, seed)) + + def test_larger_ensemble_converges(self): + for kind in (ORSet, PNCounter, LWWRegister): + for seed in (0, 1, 2): + self.assertTrue( + check_convergence(kind, seed, n=5, ops=150, rounds=6), + "%s diverged with 5 replicas at seed %d" % (kind.__name__, seed), + ) + + +class TestMutantsBreakNamedLaws(unittest.TestCase): + """Each mutant is tied to the specific law it violates.""" + + def test_sum_merge_is_not_idempotent(self): + a = PNCounterSumMerge("a") + a.increment(3) + self.assertNotEqual(a.merge(a).state_key(), a.state_key()) + self.assertEqual(a.merge(a).value(), 6) + + def test_noop_merge_is_not_commutative(self): + a, b = ORSetNoOpMerge("a"), ORSetNoOpMerge("b") + a.add("x") + b.add("y") + self.assertNotEqual(a.merge(b).state_key(), b.merge(a).state_key()) + + def test_last_argument_wins_is_not_commutative(self): + a, b = LWWLastArgumentWins("a"), LWWLastArgumentWins("b") + a.set("A", 1) + b.set("B", 2) + self.assertNotEqual(a.merge(b).value_key(), b.merge(a).value_key()) + + +class TestMutationTeeth(unittest.TestCase): + """If a broken merge survives the check, the check is decoration.""" + + MUTANTS = (ORSetDropsForeignTombstones, PNCounterDropsDecrements, + LWWLastArgumentWins, ORSetNoOpMerge, PNCounterSumMerge) + + def test_broken_merges_are_caught(self): + for mutant in self.MUTANTS: + hits = mutant_detected(mutant) + self.assertTrue(hits, "%s survived every seed -- check has no teeth" + % mutant.__name__) + for seed in hits: + self.assertFalse(check_convergence(mutant, seed)) + + def test_real_classes_pass_the_same_seeds(self): + pairs = ((ORSet, ORSetDropsForeignTombstones), + (PNCounter, PNCounterDropsDecrements), + (LWWRegister, LWWLastArgumentWins), + (ORSet, ORSetNoOpMerge), + (PNCounter, PNCounterSumMerge)) + for real, mutant in pairs: + hits = mutant_detected(mutant) + self.assertTrue(hits) + for seed in hits: + self.assertTrue(check_convergence(real, seed), + "%s failed where its mutant was caught (seed %d)" + % (real.__name__, seed)) + + def test_orset_mutant_is_semantically_wrong_not_just_state_wrong(self): + hits = mutant_detected(ORSetDropsForeignTombstones) + self.assertTrue(hits) + wrong_value = False + for seed in hits: + d = check_convergence_detail(ORSetDropsForeignTombstones, seed) + if not d["value_ok"]: + wrong_value = True + self.assertTrue(wrong_value, "OR-Set mutant only differed in bookkeeping") + + +if __name__ == "__main__": + unittest.main(verbosity=2 if "-v" in sys.argv else 1)