故障切换是一把演练过的开关

"""Failover drill: single-primary switch with health and lag guards (stdlib only)."""
from dataclasses import dataclass

MAX_LAG = 128  # Synthetic units, not seconds or a zero-data-loss guarantee.

class FailoverRefused(Exception):
    pass

@dataclass
class Node:
    name: str
    healthy: bool = True
    lag: int = 0
    is_primary: bool = False

class Cluster:
    def __init__(self):
        self.nodes = {"a": Node("a", is_primary=True), "b": Node("b")}
        self.fenced: set[str] = set()

    @property
    def primary(self) -> str:
        primaries = [n for n in self.nodes.values() if n.is_primary]
        assert len(primaries) == 1, f"must keep exactly one primary, saw {[p.name for p in primaries]}"
        return primaries[0].name

    def can_write(self, name: str) -> bool:
        node = self.nodes[name]
        return node.is_primary and node.healthy and name not in self.fenced

    def health_check(self, name: str) -> bool:
        node = self.nodes[name]
        return node.healthy and node.lag <= MAX_LAG

    def fence(self, name: str) -> None:
        self.fenced.add(name)
        assert not self.can_write(name)

    def failover(self, target: str, *, fence_old: bool) -> str:
        old = self.primary
        if target == old:
            raise FailoverRefused("target is already primary")
        if target not in self.nodes:
            raise FailoverRefused(f"unknown target {target}")
        candidate = self.nodes[target]
        if not fence_old or old not in self.fenced:
            raise FailoverRefused(f"refuse: old primary {old} is not fenced; fence first to avoid split brain")
        if not candidate.healthy:
            raise FailoverRefused(f"refuse: standby {target} is unhealthy")
        if candidate.lag > MAX_LAG:
            raise FailoverRefused(f"refuse: standby {target} lags by {candidate.lag} (> {MAX_LAG})")
        self.nodes[old].is_primary = False
        # Model explicit reactivation after the old writer is fenced.
        self.fenced.discard(target)
        candidate.is_primary = True
        assert self.primary == target and self.can_write(target)
        assert sum(self.can_write(name) for name in self.nodes) <= 1
        return target

def expect_refused(fn, label: str) -> None:
    try:
        fn()
    except FailoverRefused as exc:
        print(f"refused as expected ({label}): {exc}")
        return
    raise AssertionError(f"should have refused: {label}")

def main() -> None:
    # 1. Clean drill: healthy standby, caught-up replica, old primary fenced.
    cluster = Cluster()
    assert cluster.primary == "a"
    assert cluster.health_check("b")
    cluster.fence("a")
    assert cluster.failover("b", fence_old=True) == "b"
    print("clean switch: a -> b")

    # 2. Refuse lag beyond this policy; even accepted positive lag may lose writes.
    cluster = Cluster()
    cluster.nodes["b"].lag = 900
    assert not cluster.health_check("b")
    cluster.fence("a")
    expect_refused(lambda: cluster.failover("b", fence_old=True), "stale replica")
    assert cluster.primary == "a"

    # 3. Unhealthy standby must not promote.
    cluster.nodes["b"].lag = 0
    cluster.nodes["b"].healthy = False
    expect_refused(lambda: cluster.failover("b", fence_old=True), "unhealthy standby")
    assert cluster.primary == "a"

    # 4. Unfenced old primary must block promotion (split-brain guard).
    cluster.nodes["b"].healthy = True
    unfenced = Cluster()  # nobody fenced yet
    expect_refused(lambda: unfenced.failover("b", fence_old=False), "split brain")
    expect_refused(lambda: unfenced.failover("b", fence_old=True), "split brain without fence record")
    assert unfenced.primary == "a"

    # 5. Failback: resynchronize the candidate, then fence the CURRENT writer.
    cluster = Cluster()
    cluster.fence("a")
    cluster.failover("b", fence_old=True)
    cluster.nodes["a"].healthy = True
    cluster.nodes["a"].lag = 0
    cluster.fence("b")
    assert cluster.failover("a", fence_old=True) == "a"
    print("failback: b -> a")

    print("PASS: at most one writable node; 4 refusals + clean switch + failback passed")

if __name__ == "__main__":
    main()