#!/usr/bin/env python3
"""DDIA Ch. 10 -- a Lamport clock does not order operations on nodes that never
talk, and that gap leaks private data.

Two independent shards, each with its OWN Lamport clock:

  - the ACCOUNTS shard, stamping account writes (create, rename, set-private...);
  - the PHOTOS shard, stamping photo writes (upload, caption, delete...).

A Lamport timestamp is just a per-node counter, bumped on every local event and,
crucially, raised to `max(local, incoming)+1` ONLY when a message arrives from
another node. The whole ordering guarantee -- "if a happened-before b then
ts(a) < ts(b)" -- is built out of exactly two things: process order on one node,
and the +1 across a message. Take away the message and there is no edge.

A user does two things, in this real-world order:
    (1) set their account PRIVATE   -> a write on the accounts shard
    (2) upload a photo              -> a write on the photos shard
The shards exchange NO message for these two ops. So the user's real "(1) then
(2)" is invisible to Lamport: to the clocks these are CONCURRENT events. The
photos shard has simply seen fewer local events, so its counter is BEHIND -- and
the upload (op 2) gets a LOWER timestamp than set-private (op 1). The order
inverts.

Now a viewer does an MVCC snapshot read at a timestamp BETWEEN the two: writes
with ts <= T are visible, the rest are not. There is a T where the upload IS
visible but set-private is NOT: the viewer sees a still-PUBLIC account holding a
photo that only exists because the account was already private. The photo leaks.

THE FIX: route BOTH writes through a single LINEARIZABLE id generator -- one
monotonic source handing out ids in real order. set-private takes id k, the
later upload takes id k+1 > k. Now no snapshot can ever show the photo without
first showing the account private. The leak closes.

Deterministic: the shards' background histories are drawn from a seeded RNG, and
the two operations run in a fixed order, so the transcript reproduces exactly.
"""
import random

SEED = 7  # seeds the shards' independent background histories


# --- A Lamport clock: a per-node counter, +1 per local event -----------------------

class LamportClock:
    """One node's logical clock. `local_event()` bumps the counter for a write
    that originates here. `receive(incoming)` is the ONLY way another node's time
    enters: it lifts the counter to max(local, incoming)+1. No receive() call is
    made between the two shards for the account/photo ops -- that missing call is
    the entire bug."""

    def __init__(self, node):
        self.node = node
        self.counter = 0

    def local_event(self):
        self.counter += 1
        return self.counter

    def receive(self, incoming):
        self.counter = max(self.counter, incoming) + 1
        return self.counter


# --- A shard: a Lamport clock plus a write log ------------------------------------

class Shard:
    """A storage shard. Every write appends (timestamp, key, value) to a log.
    A snapshot read at T replays only the writes with timestamp <= T."""

    def __init__(self, name):
        self.name = name
        self.clock = LamportClock(name)
        self.log = []  # list of (ts, key, value)

    def write(self, key, value):
        ts = self.clock.local_event()
        self.log.append((ts, key, value))
        return ts

    def snapshot(self, T, key, default):
        """Value of `key` as of snapshot timestamp T: the newest write with
        ts <= T, or `default` if no such write exists."""
        latest = default
        for ts, k, v in self.log:
            if k == key and ts <= T:
                latest = v
        return latest


def burn_background_history(shard, n):
    """Simulate `n` unrelated prior writes on a shard, each a local event, so its
    Lamport counter arrives at the account/photo op already advanced by `n`."""
    for i in range(n):
        shard.write("_bg", i)


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


def main():
    rng = random.Random(SEED)

    line("=" * 72)
    line("SETUP -- two shards, each with its own independent Lamport clock")
    line("=" * 72)

    accounts = Shard("accounts")
    photos = Shard("photos")

    # Each shard has its own unrelated traffic before our user acts. The accounts
    # shard happens to be busier, so its clock is further ahead. Drawn from a
    # seeded RNG -- the inversion EMERGES from independent histories, not a hardcode.
    acc_bg = rng.randint(5, 8)
    pho_bg = rng.randint(1, 3)
    burn_background_history(accounts, acc_bg)
    burn_background_history(photos, pho_bg)

    line(f"    accounts shard: {acc_bg} prior local writes  -> Lamport counter now {accounts.clock.counter}")
    line(f"    photos   shard: {pho_bg} prior local writes  -> Lamport counter now {photos.clock.counter}")
    line("    the shards exchange NO message -- neither clock has ever seen the other's")
    line()

    line("=" * 72)
    line("STEP 1 -- the user acts: set PRIVATE, then upload a photo")
    line("=" * 72)
    line("    real-world order the user performed:")
    line("      (1) accounts.set_private   THEN   (2) photos.upload")
    line()

    ts_private = accounts.write("visibility", "private")
    ts_upload = photos.write("photo:sunset", "exists")

    line(f"    (1) set_private  on accounts shard -> Lamport ts = {ts_private}")
    line(f"    (2) upload_photo on photos   shard -> Lamport ts = {ts_upload}")
    line()
    line(f"    INVERSION: the user did (1) THEN (2), but ts(upload)={ts_upload} < ts(set_private)={ts_private}.")
    line("    No message linked the shards, so Lamport has no edge between the two ops:")
    line("    to the clocks they are CONCURRENT, and the busier accounts shard stamped")
    line("    the EARLIER real event with the LARGER number.")
    line()

    line("=" * 72)
    line("STEP 2 -- a viewer's MVCC snapshot read BETWEEN the two timestamps")
    line("=" * 72)
    T = (ts_upload + ts_private) // 2
    line(f"    snapshot timestamp T = {T}   (chosen: ts_upload={ts_upload} <= T < ts_private={ts_private})")
    line("    a write is visible in the snapshot iff its timestamp <= T")
    line()

    acct_private = accounts.snapshot(T, "visibility", default="public") == "private"
    photo_exists = photos.snapshot(T, "photo:sunset", default=None) == "exists"

    line(f"    accounts.visibility as of T : {'private' if acct_private else 'PUBLIC'}   "
         f"(set_private ts={ts_private} > T, so NOT yet visible)")
    line(f"    photos.photo:sunset as of T : {'EXISTS' if photo_exists else 'absent'}   "
         f"(upload ts={ts_upload} <= T, so visible)")
    line()
    if photo_exists and not acct_private:
        line("    >>> LEAK: the snapshot shows a PUBLIC account that already holds the photo.")
        line("    >>> The viewer sees a photo that only exists because the account was")
        line("    >>> ALREADY private -- data they were never supposed to see.")
    line()

    line("=" * 72)
    line("STEP 3 -- the fix: one LINEARIZABLE id generator for both writes")
    line("=" * 72)
    line("    both shards now take their write id from a single monotonic sequencer,")
    line("    so ids are handed out in real order regardless of which shard writes.")
    line()

    # The one sequencer would have stamped every prior event too, so it continues
    # from the total real traffic so far rather than resetting to zero.
    next_id = acc_bg + pho_bg

    def linearizable_id():
        nonlocal next_id
        next_id += 1
        return next_id

    # The user repeats the SAME real-world sequence, now stamped by the sequencer.
    fixed_private = linearizable_id()   # (1) set_private
    fixed_upload = linearizable_id()    # (2) upload, provably later

    line(f"    (1) set_private  -> linearizable id = {fixed_private}")
    line(f"    (2) upload_photo -> linearizable id = {fixed_upload}")
    line(f"    ORDER RESTORED: id(upload)={fixed_upload} > id(set_private)={fixed_private}, matching real order.")
    line()

    line("=" * 72)
    line("STEP 4 -- no snapshot can leak now: scan every boundary")
    line("=" * 72)
    fixed_log = {
        fixed_private: ("accounts.visibility", "private"),
        fixed_upload: ("photos.photo:sunset", "exists"),
    }
    leaked = False
    for T in range(0, fixed_upload + 2):
        priv = T >= fixed_private
        photo = T >= fixed_upload
        status = "leak" if (photo and not priv) else "safe"
        if status == "leak":
            leaked = True
        line(f"    T={T}: account={'private' if priv else 'public':7s} "
             f"photo={'exists' if photo else 'absent':6s} -> {status}")
    line()
    if not leaked:
        line("    >>> No snapshot shows photo-without-private. The photo can only become")
        line("    >>> visible at T >= id(upload), and by then set_private (a lower id) is")
        line("    >>> ALREADY visible. The leak is closed.")
    line()

    line("=" * 72)
    line("BOTTOM LINE")
    line("=" * 72)
    line(f"    Lamport, two independent clocks : upload ts={ts_upload} < set_private ts={ts_private}  -> order INVERTED, leak at T in [{ts_upload},{ts_private-1}]")
    line(f"    one linearizable id generator   : upload id={fixed_upload} > set_private id={fixed_private}  -> order CORRECT, no leaking snapshot")
    line("    A logical clock orders events only along message chains. Two ops on")
    line("    nodes that never exchanged a message are unordered -- only a shared")
    line("    linearizable sequencer imposes the real order they happened in.")


if __name__ == "__main__":
    main()