"""Multi-version concurrency control in one file.

Every key holds a *chain* of versions, newest first. Each version is stamped
with the commit id that created it (xmin) and the commit id that superseded it
(xmax, or None while it is still current). A transaction takes a *snapshot* --
a single integer, the newest commit id in existence when it began -- and from
then on it reads the past as it stood at that instant.

That makes visibility arithmetic, not locking, which is the famous half of
MVCC: readers never block writers. The half this toy is built to show is that
the same arithmetic decides what garbage collection may reclaim, and there the
readers block *something* after all.

There is no clock and no RNG anywhere in this file. Time IS the commit
counter. That is not a testability hack bolted on for the demo; it is what
MVCC actually does -- it replaces wall-clock ordering with an integer, so the
output is byte-identical on every run by construction.
"""

HORIZON = "horizon"  # reclaim if the version died at or before the oldest snapshot
PRECISE = "precise"  # reclaim if no living or future snapshot can see it


def visible(version, snap):
    """The entire visibility rule. One line, two comparisons, no locks."""
    return version.xmin <= snap and (version.xmax is None or version.xmax > snap)


def retain(version, rule, horizon, snaps):
    """Should a sweep keep this version? The two collectors differ only here."""
    if rule == HORIZON:
        # Half of `visible`: the death test only. A version whose xmin is still
        # in the future has to be kept regardless -- some later snapshot will
        # want it -- so xmin is never consulted. One comparison per version,
        # however many transactions are running.
        return version.xmax is None or version.xmax > horizon
    # The exact test: keep it only if somebody can still see it. Costs one
    # `visible` call per live snapshot per version.
    return any(visible(version, snap) for snap in snaps)


class Version:
    """One immutable value, alive for the commit-id interval [xmin, xmax)."""

    __slots__ = ("value", "xmin", "xmax")

    def __init__(self, value, xmin, xmax=None):
        self.value = value
        self.xmin = xmin
        self.xmax = xmax

    def __repr__(self):
        end = "inf" if self.xmax is None else self.xmax
        return f"<{self.value} [{self.xmin},{end})>"


class Txn:
    """A transaction: an id, the snapshot it froze at, and its buffered writes."""

    __slots__ = ("xid", "snapshot", "writes")

    def __init__(self, xid, snapshot):
        self.xid = xid
        self.snapshot = snapshot
        self.writes = {}


class Store:
    """A key-value store where nothing is ever overwritten, only superseded."""

    def __init__(self):
        self.chains = {}     # key -> list[Version], newest first
        self.live = {}       # xid -> Txn, the transactions still running
        self.clock = 0       # the newest commit id ever issued
        self.next_xid = 1

    # ---- transaction lifecycle ------------------------------------------

    def begin(self):
        txn = Txn(self.next_xid, self.clock)
        self.next_xid += 1
        self.live[txn.xid] = txn
        return txn

    def commit(self, txn):
        """Publish the buffered writes under one fresh commit id."""
        del self.live[txn.xid]
        if not txn.writes:
            return self.clock  # read-only: no commit id is burned
        self.clock += 1
        for key, value in txn.writes.items():
            chain = self.chains.setdefault(key, [])
            if chain and chain[0].xmax is None:
                chain[0].xmax = self.clock  # the previous head dies here
            chain.insert(0, Version(value, self.clock))
        return self.clock

    def abort(self, txn):
        """Throw the buffer away. Nothing was ever published, so nothing to undo."""
        del self.live[txn.xid]
        txn.writes.clear()

    # ---- reads and writes -----------------------------------------------

    def write(self, txn, key, value):
        txn.writes[key] = value

    def read(self, txn, key):
        value, _hops = self.probe(txn, key)
        return value

    def probe(self, txn, key):
        """read(), but also reporting how many versions had to be skipped."""
        if key in txn.writes:
            return txn.writes[key], 0  # a transaction sees its own writes
        chain = self.chains.get(key, ())
        for hops, version in enumerate(chain, start=1):
            if visible(version, txn.snapshot):
                return version.value, hops
        return None, len(chain)

    # ---- garbage collection ---------------------------------------------

    def horizon(self):
        """The oldest snapshot anyone still holds; the clock if nobody is running."""
        return min((t.snapshot for t in self.live.values()), default=self.clock)

    def snapshots(self):
        """Every snapshot that can still be used: the living ones, plus the one
        a transaction beginning right now would get."""
        return sorted({t.snapshot for t in self.live.values()} | {self.clock})

    def vacuum(self, rule, dry_run=False):
        """Sweep every chain under `rule`; return how many versions it frees."""
        # Both thresholds are read once, before the sweep. That is what makes
        # the horizon rule one comparison per version rather than one scan of
        # the live transactions per version.
        horizon, snaps = self.horizon(), self.snapshots()
        reclaimed = 0
        for key, chain in self.chains.items():
            keep = [v for v in chain if retain(v, rule, horizon, snaps)]
            reclaimed += len(chain) - len(keep)
            if not dry_run:
                self.chains[key] = keep
        return reclaimed

    # ---- inspection ------------------------------------------------------

    def total_versions(self):
        return sum(len(chain) for chain in self.chains.values())