#!/usr/bin/env python3
"""DDIA Ch. 9 -- Fencing tokens stop a zombie leaseholder from corrupting storage.

Reproduces the HBase-style split-brain bug from DDIA Figures 9-4/9-5/9-6.

A lock service grants a lease (and a fencing token) to client1. client1 then
suffers a long GC pause -- modeled explicitly here as "its lease expires while
it is paused". The lock service legitimately hands the lease to client2. When
client1 wakes it STILL believes it is the leader and issues a write. That is
split brain: two clients each think they hold the lock.

  - Without fencing, storage accepts client1's stale write and is corrupted.
  - With fencing, storage remembers the highest token it has accepted and
    rejects any write carrying a lower token, so the zombie's write bounces.

Nothing here uses wall-clock time or real threads: the pause and the message
ordering are modeled step by step, so the transcript reproduces exactly.
"""


class LockService:
    """Grants leases and stamps each grant with a monotonically increasing token."""

    def __init__(self, first_token=33):
        self._next_token = first_token
        self._holder = None

    def acquire(self, client):
        token = self._next_token
        self._next_token += 1
        self._holder = client
        return token


class StorageService:
    """The shared resource. In fenced mode it guards writes with a token check.

    max_token is the highest fencing token it has ever accepted. A write whose
    token is below that is a straggler from a client that has already lost the
    lease, so it is rejected. This is the same shape as compare-and-set: the
    resource itself checks a monotonic guard instead of trusting the client.
    """

    def __init__(self, fenced):
        self.fenced = fenced
        self.data = None
        self.max_token = 0

    def write(self, client, data, token=None):
        if self.fenced and token < self.max_token:
            return (False, "token %d < %d" % (token, self.max_token))
        if self.fenced:
            self.max_token = token
        self.data = data
        return (True, None)


def show_write(storage, client, data, token, fenced):
    tok = ("token %d" % token) if fenced else "(no token)"
    ok, reason = storage.write(client, data, token=(token if fenced else None))
    if ok:
        note = "storage now holds %r  [max_token=%d]" % (data, storage.max_token) if fenced \
            else "storage now holds %r" % (data,)
        print("    %s writes %r  %-9s -> ACCEPTED    %s" % (client, data, tok, note))
    else:
        print("    %s writes %r  %-9s -> REJECTED    %s" % (client, data, tok, reason))
    return ok


def rule():
    print("=" * 68)


# ---------------------------------------------------------------------------
# Scenario 1: no fencing -- the zombie corrupts storage.
# ---------------------------------------------------------------------------
def scenario_unfenced():
    rule()
    print("SCENARIO 1 -- no fencing token: the zombie write is accepted")
    rule()
    lock = LockService(first_token=33)
    storage = StorageService(fenced=False)

    t1 = lock.acquire("client1")
    print("  client1 acquires the lease            (would-be token %d)" % t1)
    print("  client1 enters a long GC pause ........ (frozen; cannot act)")
    print("  ... while client1 is paused, its lease EXPIRES ...")
    t2 = lock.acquire("client2")
    print("  client2 acquires the lease            (would-be token %d)" % t2)
    print()
    print("  writes reach unfenced storage:")
    show_write(storage, "client2", "client2 data", t2, fenced=False)
    print("  ... client1 WAKES, still believing it is the leader ...")
    show_write(storage, "client1", "client1 stale data", t1, fenced=False)
    print()
    print("  FINAL storage contents: %r" % storage.data)
    print("  >> CORRUPTED: the zombie's stale write overwrote the live one.")
    print("     Two clients each 'held' the lock; storage could not tell them apart.")
    print()


# ---------------------------------------------------------------------------
# Scenario 2: fencing on -- the stale token is rejected.
# ---------------------------------------------------------------------------
def scenario_fenced():
    rule()
    print("SCENARIO 2 -- fencing on: storage rejects the lower token")
    rule()
    lock = LockService(first_token=33)
    storage = StorageService(fenced=True)

    t1 = lock.acquire("client1")
    print("  client1 acquires the lease            (token %d)" % t1)
    print("  client1 enters a long GC pause ........ (frozen; cannot act)")
    print("  ... while client1 is paused, its lease EXPIRES ...")
    t2 = lock.acquire("client2")
    print("  client2 acquires the lease            (token %d)" % t2)
    print()
    print("  writes reach fenced storage (it tracks the highest token seen):")
    show_write(storage, "client2", "client2 data", t2, fenced=True)
    print("  ... client1 WAKES, still believing it is the leader ...")
    show_write(storage, "client1", "client1 stale data", t1, fenced=True)
    print()
    print("  FINAL storage contents: %r" % storage.data)
    print("  >> SAFE: token 33 < 34, so the zombie's write bounced off the guard.")
    print("     No error on client1's side was needed; the RESOURCE said no.")
    print()


# ---------------------------------------------------------------------------
# Scenario 3: a delayed in-flight request, same guard catches it (Figure 9-5).
# ---------------------------------------------------------------------------
def scenario_delayed():
    rule()
    print("SCENARIO 3 -- fencing on: a delayed in-flight write is fenced too")
    rule()
    lock = LockService(first_token=33)
    storage = StorageService(fenced=True)

    t1 = lock.acquire("client1")
    print("  client1 acquires the lease            (token %d)" % t1)
    print("  client1 SENDS a write, then its lease expires --")
    print("      the packet is delayed in the network, still in flight.")
    t2 = lock.acquire("client2")
    print("  client2 acquires the lease            (token %d)" % t2)
    print()
    print("  client2's write lands first; client1's delayed packet arrives after:")
    show_write(storage, "client2", "client2 data", t2, fenced=True)
    print("  ... the delayed token-33 packet from client1 finally arrives ...")
    show_write(storage, "client1", "client1 delayed data", t1, fenced=True)
    print()
    print("  FINAL storage contents: %r" % storage.data)
    print("  >> SAFE: the guard is on the TOKEN, not on who is 'current' -- an")
    print("     old-enough token loses no matter when its packet shows up.")
    print()


if __name__ == "__main__":
    print("Fencing tokens vs a zombie leaseholder (DDIA Figures 9-4/9-5/9-6)")
    print("Tokens are handed out monotonically: 33, then 34.")
    print()
    scenario_unfenced()
    scenario_fenced()
    scenario_delayed()
    rule()
    print("Summary")
    rule()
    print("  unfenced: client1's stale token-33 write ACCEPTED  -> storage corrupted")
    print("  fenced:   client1's stale token-33 write REJECTED  -> storage safe (holds token-34 data)")
    print("  delayed:  client1's delayed token-33 write REJECTED -> same guard, same outcome")