"""A quorum-replicated register, and the write it reported it did not do.

N replicas, one versioned value per key, tunable W and R, read-repair on the
read path -- Dynamo's shape. No threads, no sockets, no wall clock, no RNG:
*which* replicas a request reaches is an explicit list of ids, so a partial
write can be staged on purpose and a whole run is a pure function of its
schedule.

Nothing here is dishonest. `write` counts acks and compares them against W,
which is exactly what it promised to do. The trouble is that FAILED is a
statement about the acks, not about the replicas -- there is no undo in this
protocol -- and `read`'s repair step is the mechanism that promotes a failed
write's leftovers into the value everybody sees forever. `read_sets` is the
audit that proves the "forever": it re-derives, over every distinct R-sized
read set, what the cluster would answer.
"""

from itertools import combinations


class Replica:
    """One replica. Per key it holds a single `(version, value)` pair, and it
    resolves conflicts by last-write-wins on the integer version."""

    def __init__(self, rid):
        self.id = rid
        self.store = {}
        self.applied = []       # audit log: every put that actually landed

    def get(self, key):
        return self.store.get(key)

    def put(self, key, version, value, why):
        """THE RATCHET, and the load-bearing line in this file. A replica
        never moves to an older version, so once v2 is on a replica only v3
        displaces it: a partial write cannot be erased by writing around it,
        and read-repair is safe to run from any replica in any order.

        The `=` half of `>=` is the cheaper half -- it only makes a repeated
        put idempotent, which keeps `repaired` honest about what moved.
        Relaxing it to `>` changes that list and nothing else; see
        commentary 5.1."""
        cur = self.store.get(key)
        if cur is not None and cur[0] >= version:
            return False
        self.store[key] = (version, value)
        self.applied.append(dict(key=key, version=version, value=value, why=why))
        return True

    def force(self, key, version, value, why):
        """Write past the ratchet. Nothing in the protocol calls this; only
        `Cluster.rollback` does, which is the point -- undoing a partial write
        means violating the rule that keeps the replicas convergent."""
        self.store[key] = (version, value)
        self.applied.append(dict(key=key, version=version, value=value, why=why))
        return True


class Cluster:
    """N replicas plus the coordinator logic a client talks to. W and R are
    parameters rather than constants so the demo can sweep all N*N pairs and
    show that the W+R > N line is not where the behaviour changes."""

    def __init__(self, n, w, r, repair=True):
        self.replicas = [Replica(i) for i in range(1, n + 1)]
        self.n, self.w, self.r, self.repair = n, w, r, repair
        self.told = []          # audit log: the verdict handed to each client

    def rep(self, rid):
        return self.replicas[rid - 1]

    def write(self, key, version, value, reach):
        """`reach` is the list of replica ids the request actually got to;
        every other replica is partitioned, down, or just slow, which are the
        same thing to a coordinator holding a timeout.

        Note what is missing: there is no else-branch. When the ack count
        falls short of W the client is told FAILED and the replicas in `reach`
        keep the value anyway. A quorum register has no abort path, because
        aborting would need agreement, which is the thing it declined to buy.
        """
        acked = [i for i in reach if self.rep(i).put(key, version, value, "write")]
        ok = len(acked) >= self.w
        self.told.append(dict(key=key, version=version, value=value, ok=ok,
                              acked=sorted(acked), reached=sorted(reach)))
        return dict(ok=ok, acked=sorted(acked))

    def read(self, key, contact):
        """`contact` is the list of replica ids that answer. The newest
        version among them wins, and then read-repair pushes that winner back
        to the contacted replicas that are behind.

        The repair is not a cache refresh. It is a *write*, issued by a reader,
        carrying a version the coordinator never decided to commit -- and it
        is the only line in this file that changes how many replicas hold a
        value without a client having asked for anything."""
        if len(contact) < self.r:
            return dict(ok=False, value=None, version=None, repaired=[])
        live = [(i, self.rep(i).get(key)) for i in contact]
        live = [(i, v) for i, v in live if v is not None]
        if not live:
            return dict(ok=True, value=None, version=None, repaired=[])
        best = max(v for _, v in live)
        repaired = []
        if self.repair:
            for i in contact:
                if self.rep(i).put(key, best[0], best[1], "read-repair"):
                    repaired.append(i)
        return dict(ok=True, value=best[1], version=best[0],
                    repaired=sorted(repaired))

    def rollback(self, key, acked, version, value):
        """What the coordinator would have to do to make FAILED mean failed:
        go back to the replicas that took the write and put the old value
        back. It needs `force`, and it needs the coordinator to still be
        alive. See section 7 of the commentary for what happens when it is
        not."""
        for i in acked:
            self.rep(i).force(key, version, value, "rollback")

    def anti_entropy(self, key, pairs):
        """Background repair, the other way a version spreads. Each pair of
        replicas compares its copy and both take the newer one. Real systems
        drive this off Merkle trees; `pairs` here is an explicit list so the
        pass is reproducible."""
        moved = []
        for a, b in pairs:
            va, vb = self.rep(a).get(key), self.rep(b).get(key)
            if va is None or vb is None or va[0] == vb[0]:
                continue
            hi = max(va, vb)
            self.rep(a).put(key, hi[0], hi[1], "anti-entropy")
            self.rep(b).put(key, hi[0], hi[1], "anti-entropy")
            moved.append((a, b, hi))
        return moved

    # ---- audit -----------------------------------------------------------

    def state(self, key):
        return [(rep.id, rep.get(key)) for rep in self.replicas]

    def holders(self, key, version):
        """Which replicas hold `version` or newer -- the real durability of a
        write, as opposed to the verdict its client was handed."""
        return [rep.id for rep in self.replicas
                if (rep.get(key) or (0,))[0] >= version]

    def read_sets(self, key, r=None):
        """THE PERMANENCE AUDIT. Every distinct R-sized set of replicas, and
        the value a read from that set would return. Deliberately does not go
        through `read`: it must not repair, or the act of auditing would
        change the thing being audited."""
        r = self.r if r is None else r
        out = []
        for s in combinations(range(1, self.n + 1), r):
            vs = [self.rep(i).get(key) for i in s]
            vs = [v for v in vs if v is not None]
            out.append((list(s), max(vs)[1] if vs else None))
        return out

    def count_returning(self, key, value, r=None):
        """How many of those read sets return `value`, and how many there are.
        The 6/10 -> 10/10 in section 6 is two calls to this."""
        sets = self.read_sets(key, r)
        return sum(1 for _, v in sets if v == value), len(sets)