#!/usr/bin/env python3
"""DDIA Ch. 10 -- a Hybrid Logical Clock survives a backward NTP step that
reorders plain wall-clock timestamps.

One node performs a CAUSAL CHAIN of writes: each event is a read-modify-write on
a shared register `x` (event k reads x=k-1 and writes x=k), so event k+1 provably
happens-AFTER event k. We timestamp the same chain two ways:

  (1) PLAIN PHYSICAL TIME -- read straight off a wall clock.
  (2) A HYBRID LOGICAL CLOCK (HLC) -- physical time plus a logical counter.

Partway through the chain, NTP decides the clock is fast and STEPS IT BACKWARD by
100 ms (a correction). This is modelled by a mock physical clock we can step, so
the whole run is deterministic.

The plain physical timestamps then go NON-MONOTONIC across the jump: a LATER event
carries a SMALLER timestamp than its own cause. Ordering the events by physical
timestamp therefore places an effect before the cause it read -- a causality
inversion -- and a snapshot read at a timestamp in the gap sees the effects but
not the committed cause.

The HLC applies one rule on every event:

    pt_new = max(pt_last, phys)
    counter = counter + 1   if pt_new == pt_last   (a tie or a backward step)
            = 0             otherwise               (physical time moved us forward)

and on receiving a remote timestamp it first takes the max with that too. The
(pt, counter) pair is compared lexicographically. Because pt never decreases and
the counter breaks ties, the HLC timestamps are STRICTLY MONOTONIC and preserve
the causal order -- while staying within the jump's bound of physical time.

Deterministic: the physical clock is a mock driven by a fixed script of
advance/step operations, so the transcript reproduces byte-for-byte every run.
"""


# --- A mock physical clock we can advance and STEP BACKWARD ------------------------

class MockPhysicalClock:
    """A wall clock, in milliseconds. `advance(dt)` is time passing normally.
    `step(dt)` is an external authority (NTP) resetting it -- dt may be NEGATIVE,
    which is the backward correction this exercise turns on. `now()` reads it."""

    def __init__(self, start_ms):
        self.t = start_ms

    def advance(self, dt_ms):
        self.t += dt_ms

    def step(self, dt_ms):
        self.t += dt_ms  # NTP correction; dt_ms < 0 steps the clock BACK

    def now(self):
        return self.t


# --- A Hybrid Logical Clock -------------------------------------------------------

class HLC:
    """Physical time that is never allowed to go backward. State is (pt, counter).
    On every local event we clamp pt up to physical time and, when physical time
    did NOT advance us (a tie or a backward step), bump the counter instead;
    otherwise the counter resets to 0. `receive` first takes the max with a
    remote HLC so causality carried by a message is preserved. Timestamps compare
    lexicographically as (pt, counter)."""

    def __init__(self):
        self.pt = 0
        self.counter = 0

    def local_event(self, phys):
        pt_last = self.pt
        self.pt = max(pt_last, phys)
        if self.pt == pt_last:
            self.counter += 1        # physical time didn't move forward -> break the tie
        else:
            self.counter = 0         # physical time carried us forward -> reset
        return (self.pt, self.counter)

    def receive(self, phys, remote_pt, remote_counter):
        pt_last = self.pt
        self.pt = max(pt_last, remote_pt, phys)
        if self.pt == pt_last == remote_pt:
            self.counter = max(self.counter, remote_counter) + 1
        elif self.pt == pt_last:
            self.counter += 1
        elif self.pt == remote_pt:
            self.counter = remote_counter + 1
        else:
            self.counter = 0
        return (self.pt, self.counter)


def fmt_hlc(ts):
    return f"({ts[0]}, c{ts[1]})"


def line(s=""):
    print(s)


def main():
    line("=" * 74)
    line("SETUP -- one node, a causal chain of writes, timestamped two ways")
    line("=" * 74)
    line("    Each event is a read-modify-write on register x:")
    line("      event k reads x = k-1 and writes x = k, so event k+1 happens-AFTER k.")
    line("    Physical clock starts at 1000 ms and normally advances +10 ms per event.")
    line("    Between event 4 and event 5, NTP STEPS THE CLOCK BACK 100 ms.")
    line()

    clock = MockPhysicalClock(start_ms=1000)
    hlc = HLC()

    # Fixed script: (label, advance-before-this-event, is_ntp_step_here)
    # Events 1..7; the -100 ms step lands just before event 5.
    events = []  # (name, phys_ts, hlc_ts)

    def do_event(name, advance_ms, step_ms=0):
        if step_ms:
            clock.step(step_ms)
        else:
            clock.advance(advance_ms)
        phys = clock.now()
        ts = hlc.local_event(phys)
        events.append((name, phys, ts))
        return phys, ts

    # event 1 lands at the start value 1000 (advance 0), then +10 each.
    do_event("e1  x:=1", 0)
    do_event("e2  x:=2", 10)
    do_event("e3  x:=3", 10)
    do_event("e4  x:=4", 10)
    # NTP correction: the clock was 100 ms fast, step it back.
    do_event("e5  x:=5", 0, step_ms=-100 + 10)   # net: back 100, then the 10 ms of work
    do_event("e6  x:=6", 10)
    do_event("e7  x:=7", 10)

    line("=" * 74)
    line("STEP 1 -- the physical timestamps of the causal chain")
    line("=" * 74)
    for name, phys, ts in events:
        line(f"    {name}   physical_ts = {phys} ms")
    line()
    line("    The NTP step lands between e4 and e5:")
    e4 = events[3]
    e5 = events[4]
    line(f"      e4 committed at physical_ts = {e4[1]} ms")
    line(f"      e5 committed at physical_ts = {e5[1]} ms   (clock was stepped BACK 100 ms)")
    line(f"    e5 happens-AFTER e4 (it read x=4, wrote x=5), yet {e5[1]} < {e4[1]}:")
    line("    a LATER event carries a SMALLER physical timestamp -- NON-MONOTONIC.")
    line()

    line("=" * 74)
    line("STEP 2 -- the failure: order the events by physical timestamp")
    line("=" * 74)
    by_phys = sorted(events, key=lambda e: e[1])
    line("    sorted by physical_ts (what a timestamp-ordered log / snapshot would do):")
    for name, phys, ts in by_phys:
        line(f"      physical_ts = {phys} ms   {name}")
    line()
    order = [e[0].split()[0] for e in by_phys]
    line(f"    causal order actually performed : e1 -> e2 -> e3 -> e4 -> e5 -> e6 -> e7")
    line(f"    order by physical timestamp     : {' -> '.join(order)}")
    line("    e5, e6, e7 sort BEFORE e4 -- their cause. Sorting by physical time")
    line("    places the effect ahead of the write it read: causality INVERTED.")
    line()
    # snapshot failure: a reader takes a snapshot at a ts in the gap
    T = 1025
    visible = [e[0].split()[0] for e in events if e[1] <= T]
    line(f"    a snapshot read at physical_ts T = {T} ms (writes with ts <= T are visible):")
    line(f"      visible: {', '.join(visible)}")
    line("      -> e5, e6, e7 are visible but e4 (which they causally depend on) is NOT.")
    line("      The snapshot shows x=7 while the write that set x=4 is missing: a state")
    line("      that never existed. A committed earlier-in-causal-order write is lost.")
    line()

    line("=" * 74)
    line("STEP 3 -- the same chain, timestamped by the HLC")
    line("=" * 74)
    for name, phys, ts in events:
        line(f"    {name}   physical_ts = {phys:>4} ms   hlc = {fmt_hlc(ts)}")
    line()
    line("    Across the jump the HLC does NOT go back:")
    line(f"      e4 hlc = {fmt_hlc(e4[2])}")
    line(f"      e5 hlc = {fmt_hlc(e5[2])}   (pt held at {e5[2][0]}, counter bumped 0 -> 1)")
    line("    physical time tried to regress, so pt stayed put and the counter absorbed")
    line("    the step. e5, e6, e7 keep climbing on the counter: (1030,c1) < (1030,c2) < (1030,c3).")
    line()
    hlcs = [e[2] for e in events]
    strictly_monotonic = all(hlcs[i] < hlcs[i + 1] for i in range(len(hlcs) - 1))
    line(f"    HLC timestamps strictly increasing (lexicographic (pt, counter))? {strictly_monotonic}")
    line("    Ordering the events by HLC reproduces the exact causal order e1..e7.")
    line()

    line("=" * 74)
    line("STEP 4 -- receive(): causality carried across a message")
    line("=" * 74)
    line("    A peer node B, whose clock is AHEAD, sends a message stamped with its HLC.")
    line("    Our node applies receive(): it takes the max with the remote HLC first,")
    line("    so the event that READ the message is ordered after it.")
    line()
    remote_pt, remote_counter = 1200, 4      # peer B is ahead of our stepped-back clock
    phys_here = clock.now()
    line(f"    our physical clock now        = {phys_here} ms")
    line(f"    message arrives from B stamped  hlc = {fmt_hlc((remote_pt, remote_counter))}  (B is ahead)")
    before = (hlc.pt, hlc.counter)
    got = hlc.receive(phys_here, remote_pt, remote_counter)
    line(f"    our HLC before receive        = {fmt_hlc(before)}")
    line(f"    our HLC after  receive        = {fmt_hlc(got)}")
    line(f"    it jumped up to B's physical part ({remote_pt}) and set counter = {got[1]}:")
    line("    strictly greater than the message's own stamp, so the receipt is ordered")
    line("    AFTER the send even though our local physical clock never read that high.")
    line()

    line("=" * 74)
    line("STEP 5 -- bounded skew: the HLC still tracks physical time")
    line("=" * 74)
    line("    Right after the backward step, before the remote message:")
    line(f"      e7 physical_ts = {e5[1] + 20} ms     e7 hlc.pt = {events[6][2][0]}")
    skew = events[6][2][0] - (e5[1] + 20)
    line(f"      skew = hlc.pt - physical_ts = {skew} ms")
    line(f"    The HLC ran at most {skew} ms ahead of the wall clock -- exactly the part of")
    line("    the 100 ms backward step not yet re-covered by real time. The counter, not")
    line("    an unbounded drift, carried the ordering; the skew is bounded by the jump.")
    line()

    line("=" * 74)
    line("BOTTOM LINE")
    line("=" * 74)
    line(f"    physical timestamps : e4={e4[1]}, e5={e5[1]}  -> e5 < e4, NON-monotonic; sort inverts causality")
    line(f"    HLC timestamps      : e4={fmt_hlc(e4[2])}, e5={fmt_hlc(e5[2])}  -> strictly increasing, causal order preserved")
    line("    'NTP-synced clocks are correctly ordered' fails the instant the clock")
    line("    steps back. The HLC's max-and-increment rule buys a monotonic total order")
    line("    from the same rough physical time -- no special hardware required.")


if __name__ == "__main__":
    main()