"""Two-phase commit, as an explicit and fully enumerable message schedule.

No threads, no sockets, no clock, no RNG. Every movement of every message is
one numbered STEP, so "crash the coordinator at step k" is a total function
over the protocol and the entire crash sweep is a `for` loop.

The step model is the load-bearing modelling decision:

    SEND(node, msg)     node hands msg to the network. Requires node alive.
    DELIVER(node, msg)  node's handler runs.           Requires node alive.

Splitting the send from the delivery is what lets a crash land *between*
"decided, and forced COMMIT to the log" and "the first COMMIT is on the
wire" -- the window the protocol's whole reputation rests on. A model where
a handler atomically emits all of its outbound messages cannot express it.
"""

from collections import deque

INIT, PREPARED, COMMITTED, ABORTED = "init", "prepared", "committed", "aborted"
C = "C"  # the coordinator's node id


class Participant:
    """A resource manager. `prepared` is the dangerous state: locks are held,
    undo/redo is forced to disk, and the right to abort unilaterally has been
    GIVEN AWAY. That surrender is the whole protocol."""

    def __init__(self, pid, vote):
        self.id = pid
        self.vote = vote            # "yes" | "no", fixed by the schedule
        self.state = INIT
        self.log = []               # forced records; these survive a crash
        self.locks = 0

    def on_prepare(self, m):
        if self.vote == "no":
            # THE line the one-NO result lives on. A NO voter never enters the
            # uncertainty period at all: it knows the outcome by itself, so it
            # aborts here and stays a live oracle for everybody else.
            self.state = ABORTED
            self.log.append("abort")
            return [dict(src=self.id, dst=C, kind="vote", vote="no")]
        self.state = PREPARED
        self.locks = 1
        self.log.append("prepare")  # forced BEFORE the reply goes out
        return [dict(src=self.id, dst=C, kind="vote", vote="yes")]

    def on_decision(self, m):
        if self.state in (COMMITTED, ABORTED):
            return []               # decisions are idempotent; retries are free
        self.state = COMMITTED if m["decision"] == "commit" else ABORTED
        self.log.append(m["decision"])
        self.locks = 0
        return [dict(src=self.id, dst=C, kind="ack")]

    def uncertain(self):
        """Prepared, and has not heard the outcome. Cannot act, cannot let go."""
        return self.state == PREPARED


class Coordinator:
    """Collects votes, decides once, and is the only node that ever knows."""

    def __init__(self, pids):
        self.id = C
        self.pids = pids
        self.state = INIT
        self.votes = {}
        self.log = []

    def start(self):
        self.state = "collecting"
        return [dict(src=C, dst=p, kind="prepare") for p in self.pids]

    def on_vote(self, m):
        if self.state != "collecting":
            return []
        self.votes[m["src"]] = m["vote"]
        if len(self.votes) < len(self.pids):
            return []               # unanimity required, so one NO is enough
        d = "commit" if all(v == "yes" for v in self.votes.values()) else "abort"
        self.state = COMMITTED if d == "commit" else ABORTED
        self.log.append(d)          # the decision record, forced to stable store
        return [dict(src=C, dst=p, kind="decision", decision=d) for p in self.pids]

    def on_ack(self, m):
        return []


class Sim:
    """The driver: owns the nodes, the event queue, and the schedule."""

    def __init__(self, votes):
        """`votes` is one entry per participant, e.g. ["yes", "no", "yes"]."""
        self.pids = list(range(1, len(votes) + 1))
        self.parts = {p: Participant(p, v) for p, v in zip(self.pids, votes)}
        self.coord = Coordinator(self.pids)
        self.nodes = {C: self.coord}
        self.nodes.update(self.parts)
        self.alive = {n: True for n in self.nodes}
        self.trace = []
        self.q = deque(("SEND", m) for m in self.coord.start())

    def step(self):
        """Execute exactly one event, and return its one-line description."""
        kind, m = self.q.popleft()
        tag = f"{m['src']}->{m['dst']} {m['kind']:8}{self._payload(m)}"
        if kind == "SEND":
            if self.alive[m["src"]]:
                self.q.append(("DELIVER", m))
                desc = f"SEND    {tag}"
            else:
                desc = f"SEND    {tag} NEVER SENT (sender dead)"
        elif self.alive[m["dst"]]:
            for out in getattr(self.nodes[m["dst"]], "on_" + m["kind"])(m):
                self.q.append(("SEND", out))
            desc = f"DELIVER {tag}"
        else:
            desc = f"DELIVER {tag} DROPPED (receiver dead)"
        self.trace.append(desc.rstrip())
        return desc.rstrip()

    @staticmethod
    def _payload(m):
        for k in ("vote", "decision"):
            if k in m:
                return " " + m[k]
        return ""

    def run(self, limit=600):
        """Drain the queue. Returns the number of steps executed."""
        n = 0
        while self.q and n < limit:
            self.step()
            n += 1
        return n

    def crash(self, node):
        """Stop the node dead. Its forced log survives; it sends and receives
        nothing ever again. Messages it already SENT stay in flight."""
        self.alive[node] = False
        self.trace.append(f"*** {node} CRASHES ***")

    # ---- what the survivors can work out with the coordinator gone ----

    def terminate(self):
        """The cooperative termination protocol (Bernstein, Hadzilacos and
        Goodman s7.4): an uncertain participant asks every live peer what it
        knows, and the group resolves if ANY peer is outside the uncertainty
        period.

        Returns "commit", "abort", or "BLOCKED" -- the last meaning every live
        participant is uncertain, so the group's uncertainty set is still
        {commit, abort} and no rule can pick between them.
        """
        live = [p for p in self.parts.values() if self.alive[p.id]]
        known = None
        for p in live:
            if p.state == COMMITTED:
                known = "commit"
            elif p.state == ABORTED and known != "commit":
                known = "abort"
            elif p.state == INIT and known is None:
                known = "abort"     # it never voted, so commit was impossible
        if known is None:
            return "BLOCKED"
        for p in live:
            p.on_decision(dict(decision=known))
        return known

    def unilateral(self, heuristic):
        """The other option: no peer query. Every uncertain participant just
        guesses `heuristic` when its timer fires. Never blocks; can corrupt."""
        for p in self.parts.values():
            if self.alive[p.id] and p.uncertain():
                p.on_decision(dict(decision=heuristic))

    def outcomes(self):
        return {p.id: p.state for p in sorted(self.parts.values(), key=lambda x: x.id)}

    def locks_held(self):
        return sum(p.locks for p in self.parts.values())


def total_steps(votes):
    """How many steps a clean, crash-free run of this vote vector takes."""
    return Sim(votes).run()


def run_crash_at(votes, k, recovery="terminate", heuristic="abort"):
    """Run the schedule, crash the coordinator immediately after step `k`,
    drain whatever is still in flight, then recover. `k=0` means it died
    before doing anything at all. `recovery="none"` freezes the cluster
    exactly as the crash left it, for inspection."""
    s = Sim(votes)
    for _ in range(k):
        if not s.q:
            break
        s.step()
    s.crash(C)
    s.run()
    if recovery == "terminate":
        return s, s.terminate()
    if recovery == "unilateral":
        s.unilateral(heuristic)
        return s, None
    if recovery == "none":
        return s, None
    raise ValueError(recovery)


def blocked_timings(votes):
    """Every crash step that leaves the survivors stuck. The headline sweep."""
    n = total_steps(votes)
    return [k for k in range(n + 1)
            if run_crash_at(votes, k)[1] == "BLOCKED"], n