LWW versus a G-counter under clock skew

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""LWW versus a G-counter under clock skew.

Three replicas, one with a clock several seconds ahead, take 1,000 increments
of a points counter and merge in random order. Compared: one integer merged
by last-writer-wins, one integer merged with max, and a G-counter. Then
decrements (PN-counter) and a "never below zero" rule.
"""
import argparse, json, random

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--increments", type=int, default=1000, help="Increments spread over the replicas.")
p.add_argument("--skew", type=float, default=3.0, help="Seconds replica C's clock is ahead.")
p.add_argument("--merge-every", type=int, default=20, help="Operations between random pairwise merges.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
random.seed(args.seed)
R = ["A", "B", "C"]
offset = {"A": 0.0, "B": 0.0, "C": args.skew}


def run(design, ops, dup=False):
    """ops: list of (replica, delta). Returns the converged value after full merges."""
    state = {r: ({"v": 0, "ts": 0.0} if design in ("lww", "max") else {"P": {}, "N": {}}) for r in R}
    now = 0.0
    def merge(a, b):                                   # b absorbs a
        sa, sb = state[a], state[b]
        if design == "lww":
            if sa["ts"] > sb["ts"]: sb.update(sa)
        elif design == "max":
            sb["v"] = max(sa["v"], sb["v"])
        else:
            for part in ("P", "N"):
                for k in set(sa[part]) | set(sb[part]):
                    sb[part][k] = max(sa[part].get(k, 0), sb[part].get(k, 0))
    for i, (r, d) in enumerate(ops):
        now += 0.01
        s = state[r]
        if design in ("lww", "max"):
            s["v"] += d; s["ts"] = now + offset[r]
        else:
            part = "P" if d > 0 else "N"
            s[part][r] = s[part].get(r, 0) + abs(d)
        if i % args.merge_every == 0:
            a, b = random.sample(R, 2); merge(a, b)
            if dup: merge(a, b)                        # a duplicated delivery
    for _ in range(3):                                  # final anti-entropy: everyone merges with everyone
        for a in R:
            for b in R:
                if a != b: merge(a, b)
    s = state["A"]
    return s["v"] if design in ("lww", "max") else sum(s["P"].values()) - sum(s["N"].values())


inc = [(random.choice(R), +1) for _ in range(args.increments)]
rows = []
for design, label in (("lww", "one integer, last writer wins"), ("max", "one integer, merged with max"), ("g", "G-counter")):
    v = run(design, inc); vd = run(design, inc, dup=True)
    rows.append({"design": label, "value": v, "lost": args.increments - v, "value_with_duplicates": vd})
    print(f"{label:<32} final {v:>5} of {args.increments}  (lost {args.increments - v:>4}; with duplicated merges: {vd})")

# who wins under LWW depends on the clock: C (ahead) overwrites the others
skewless = args.skew; offset["C"] = 0.0
v0 = run("lww", inc); offset["C"] = skewless
print(f"LWW with no skew: {v0} (with C {args.skew:g} s ahead: {rows[0]['value']})")

# decrements: a G-counter cannot take them; a PN-counter (two G-counters) can
mixed = inc + [(random.choice(R), -1) for _ in range(300)]
random.shuffle(mixed)
pn = run("g", mixed)
print(f"PN-counter with {args.increments} increments and 300 decrements: {pn} (expected {args.increments - 300})")

# "never below zero": two replicas each allow a redemption against the same balance of 100
balance, redeem = 100, 80
local = {"A": balance, "B": balance}
ok = [r for r in ("A", "B") if local[r] - redeem >= 0]   # each checks only its own copy
merged = balance - redeem * len(ok)
print(f"two replicas each approve a {redeem}-point redemption from {balance}: merged balance {merged}")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"designs": rows, "summary": [{
            "lww": rows[0]["value"], "max": rows[1]["value"], "g": rows[2]["value"], "lww_noskew": v0,
            "pn": pn, "pn_expected": args.increments - 300, "negative_balance": merged}]}}, f)