"""One staged partial write, then five sweeps.

The schedule is an adversary, and every part of it is a literal argument:
v2 reaches `reach=[3]` and nothing else, and each read names the replicas that
answer it. Nothing is randomised, so "the read that happened to touch r3" is a
line of code rather than a lucky run.

The sweeps exist because "raise W", "use W+R > N", "turn off read-repair" and
"roll it back" are the four fixes every reader proposes, and all four can be
measured instead of argued about.

  python3 demo.py
"""

from quorum import Cluster

N, W, R, KEY = 5, 3, 3, "cart"
BAR = "=" * 74


def show(c, label):
    row = "  ".join(f"r{i}=" + ("-" if v is None else f"v{v[0]}:{v[1]}")
                    for i, v in c.state(KEY))
    print(f"    {label:<24} {row}")


def staged(w=W, r=R, reach=(3,), repair=True, verbose=False):
    """v1 lands everywhere; v2 reaches only `reach`. Returns the cluster and
    the verdict the client was handed for v2."""
    c = Cluster(N, w, r, repair=repair)
    a = c.write(KEY, 1, "A", reach=list(range(1, N + 1)))
    if verbose:
        print(f"  t0  write v1='A', reaches all {N} -> acks={a['acked']}"
              f"  CLIENT TOLD: {'OK' if a['ok'] else 'FAILED'}")
        show(c, "after v1")
    b = c.write(KEY, 2, "B", reach=list(reach))
    if verbose:
        print(f"  t1  write v2='B', reaches {sorted(reach)} -> acks={b['acked']}"
              f", W={w}  CLIENT TOLD: {'OK' if b['ok'] else 'FAILED'}")
        show(c, "after the failed v2")
    return c, b


print(BAR)
print(f"  THE STAGED RUN --  N={N}  W={W}  R={R}   (W+R = {W + R} > N = {N})")
print(BAR)
c, b = staged(verbose=True)
pre = c.count_returning(KEY, "B")
print(f"    read sets returning 'B' at this point: {pre[0]}/{pre[1]}")

print()
print("  t2  read contacting [1, 2, 4] -- does not touch r3")
g = c.read(KEY, [1, 2, 4])
print(f"      client reads v{g['version']}='{g['value']}', repaired={g['repaired']}")
show(c, "unchanged")

print()
print("  t3  read contacting [2, 3, 4] -- touches r3, which took the failed v2")
g = c.read(KEY, [2, 3, 4])
print(f"      client reads v{g['version']}='{g['value']}', repaired={g['repaired']}")
show(c, "after read-repair")
post = c.count_returning(KEY, "B")

print()
print(f"  VERDICT   the client was told v2='B' {'SUCCEEDED' if b['ok'] else 'FAILED'}")
print(f"            replicas holding v2:      {c.holders(KEY, 2)}"
      f"  ({len(c.holders(KEY, 2))} of {N})")
print(f"            read sets returning 'B':  {pre[0]}/{pre[1]} before the read,"
      f"  {post[0]}/{post[1]} after it")

print()
print(BAR)
print("  ASIDE -- the same defect running backwards.  N=5 W=1 R=1")
print(BAR)
c1, b1 = staged(w=1, r=1)
n1, t1 = c1.count_returning(KEY, "B", r=1)
print(f"    v2 reaches only r3, W=1 -> CLIENT TOLD: {'OK' if b1['ok'] else 'FAILED'}")
print(f"    R=1 read sets returning 'B': {n1}/{t1}"
      f"   -- {t1 - n1} of {t1} readers still get the old value")
print("    The verdict the client hears and the state of the cluster are two")
print("    different quantities. Neither one is a lie; they just aren't the same.")

print()
print(BAR)
print("  SWEEP 1 -- does W+R > N prevent it?   v2 reaches only r3")
print(BAR)
print("     W   R  W+R>N | told   | 'B' before | 'B' after one read touching r3")
for w in range(1, N + 1):
    for r in range(1, N + 1):
        cw, bw = staged(w=w, r=r)
        before = cw.count_returning(KEY, "B")
        s = next(t for t, _ in cw.read_sets(KEY, r) if 3 in t)
        cw.read(KEY, s)
        after = cw.count_returning(KEY, "B")
        print(f"     {w}   {r}  {str(w + r > N):>5} | "
              f"{'OK' if bw['ok'] else 'FAILED':<6} |"
              f"   {f'{before[0]}/{before[1]}':>6}   |"
              f"   {f'{after[0]}/{after[1]}':>6}")

print()
print(BAR)
print("  SWEEP 2 -- does turning read-repair OFF prevent it?   N=5 W=3 R=3")
print(BAR)
READS = ([1, 2, 4], [2, 3, 4], [1, 2, 4], [1, 4, 5], [1, 2, 5])
for repair in (True, False):
    cr, _ = staged(repair=repair)
    before = cr.count_returning(KEY, "B")
    seen = [cr.read(KEY, list(s))["value"] for s in READS]
    after = cr.count_returning(KEY, "B")
    print(f"    read_repair={str(repair):<5}  sets returning 'B':"
          f" {before[0]}/{before[1]} before, {after[0]}/{after[1]} after five reads"
          f"   client saw: {' '.join(seen)}")

print()
print("  and with read-repair off, one anti-entropy pass, a pair at a time:")
ca, _ = staged(repair=False)
for pair in [(1, 3), (1, 4), (1, 5), (2, 3), (2, 5)]:
    for a, bb, hi in ca.anti_entropy(KEY, [pair]):
        n, t = ca.count_returning(KEY, "B")
        print(f"    gossip r{a} <-> r{bb}: both now v{hi[0]}='{hi[1]}'"
              f"   sets returning 'B': {n:>2}/{t}")
print(f"    replicas holding v2 after the pass: {ca.holders(KEY, 2)}")

print()
print(BAR)
print("  SWEEP 3 -- does a HIGHER W help?   reach = how many replicas v2 got to")
print(BAR)
print("    reach |" + "".join(f"{'W=' + str(w):>10}" for w in range(1, N + 1)))
for k in range(0, N + 1):
    reach = list(range(1, k + 1))
    cells = []
    for w in range(1, N + 1):
        ck, bk = staged(w=w, reach=reach)
        if reach:
            s = next(t for t, _ in ck.read_sets(KEY) if any(x in reach for x in t))
            ck.read(KEY, s)
        n, t = ck.count_returning(KEY, "B")
        survives = "B" if n == t else ("A" if n == 0 else f"{n}/{t}B")
        cells.append(f"{'OK' if bk['ok'] else 'FAILED'}/{survives}")
    print(f"    {k:>5} |" + "".join(f"{x:>10}" for x in cells))
print("    told / what every R=3 read returns after one read touching the residue")

print()
print(BAR)
print("  SWEEP 4 -- rollback: can the coordinator take the failed write back?")
print(BAR)
for crash in (False, True):
    cb, bb = staged()
    if not crash:
        cb.rollback(KEY, bb["acked"], 1, "A")
    cb.read(KEY, [2, 3, 4])
    final = cb.read_sets(KEY)[0][1]
    print(f"    coordinator crashes before rolling back = {str(crash):<5}"
          f" -> final value {final!r}, replicas on v2: {len(cb.holders(KEY, 2))}")
print("    Rollback needs Replica.force -- it writes past the version ratchet --")
print("    and it needs the coordinator to outlive the write it is undoing.")

print()
print(BAR)
print("  SWEEP 5 -- the two boundaries, where the effect vanishes")
print(BAR)
c0, b0 = staged(reach=())
c0.read(KEY, [2, 3, 4])
n0, t0 = c0.count_returning(KEY, "B")
print(f"    reach=[] : told {'OK' if b0['ok'] else 'FAILED'}, "
      f"sets returning 'B' after a read: {n0}/{t0}  (nothing to promote)")
c9, b9 = staged(r=1)
n9, t9 = c9.count_returning(KEY, "B", r=1)
g9 = c9.read(KEY, [3])
n8, t8 = c9.count_returning(KEY, "B", r=1)
print(f"    R=1      : the read touching r3 returns {g9['value']!r} and repairs "
      f"{g9['repaired']}")
print(f"               sets returning 'B': {n9}/{t9} before, {n8}/{t8} after"
      f"  (repair has nowhere to push)")