#!/usr/bin/env python3
"""Exact teaching state machines, not a storage implementation.
LSNs occupy fixed logical slots 10,20,...; file I/O is not performed.
All checks remain enabled with python -O. Results go only to stdout.
"""
from dataclasses import dataclass
from itertools import product, permutations
from copy import deepcopy
import json
import random


def require(test, message):
    if not test:
        raise ValueError(message)


@dataclass(frozen=True)
class Record:
    lsn: int
    tx: str
    kind: str
    prev: int
    page: str = ''
    before: int = 0
    after: int = 0
    undo_next: int = 0


class Engine:
    def __init__(self, names=('A', 'P', 'Q', 'R')):
        require(len(set(names)) == len(names), 'duplicate page identity')
        self.pages = {p: (0, 0) for p in names}  # value, pageLSN
        self.disk = self.pages.copy()
        self.records, self.index = [], {}
        self.tx, self.owner, self.dirty = {}, {}, {}
        self.stable = 0
        self.serial = 0

    @property
    def tail(self):
        return self.records[-1].lsn if self.records else 0

    def append(self, tx, kind, page='', before=0, after=0, undo_next=0):
        state = self.tx[tx]
        r = Record(self.tail + 10, tx, kind, state['last'], page, before, after, undo_next)
        self.records.append(r); self.index[r.lsn] = r; state['last'] = r.lsn
        return r

    def apply(self, r):
        require(r.page in self.pages, 'unknown page')
        require(self.pages[r.page][1] < r.lsn, 'apply must advance pageLSN')
        self.dirty.setdefault(r.page, r.lsn)
        self.pages[r.page] = (r.after, r.lsn)

    def begin(self, tx):
        require(tx not in self.tx, 'transaction identity cannot be reused')
        self.tx[tx] = {'status': 'ACTIVE', 'first': self.tail+10, 'last': 0, 'marks': []}
        return self.append(tx, 'BEGIN')

    def active(self, tx):
        require(tx in self.tx and self.tx[tx]['status'] == 'ACTIVE', 'transaction not active')
        return self.tx[tx]

    def write(self, tx, page, value):
        self.active(tx)
        require(page in self.pages and type(value) is int, 'integer page value required')
        require(page not in self.owner or self.owner[page] == tx, 'conflicting lock held')
        self.owner[page] = tx
        r = self.append(tx, 'UPDATE', page, self.pages[page][0], value)
        self.apply(r)
        return r

    def savepoint(self, tx):
        st = self.active(tx); self.serial += 1
        token = (tx, self.serial)
        st['marks'].append((token, st['last']))
        return token

    def mark_index(self, tx, token):
        st = self.active(tx)
        for i, (t, _) in enumerate(st['marks']):
            if t == token:
                return i
        raise ValueError('foreign or expired savepoint')

    def undo_record(self, tx, cursor):
        r = self.index[cursor]
        require(r.tx == tx, 'transaction chain identity')
        if r.kind == 'CLR':
            return r.undo_next, None
        if r.kind == 'UPDATE':
            clr = self.append(tx, 'CLR', r.page, after=r.before, undo_next=r.prev)
            self.apply(clr)
            return r.prev, clr
        return r.prev, None

    def rollback_to(self, tx, token):
        i = self.mark_index(tx, token); st = self.tx[tx]
        stop = st['marks'][i][1]; cursor = st['last']; visited = []; generated = []
        st['status'] = 'ROLLING_BACK'
        while cursor > stop:
            visited.append(cursor)
            nxt, clr = self.undo_record(tx, cursor)
            require(nxt < cursor, 'undo cursor must descend')
            cursor = nxt
            if clr is not None:
                generated.append(clr.lsn)
        del st['marks'][i+1:]
        st['status'] = 'ACTIVE'
        return {'visited': visited, 'new_clrs': generated}

    def release(self, tx, token):
        i = self.mark_index(tx, token)
        del self.tx[tx]['marks'][i:]

    def flush_log(self, through):
        require(through == 0 or through in self.index, 'whole log boundary required')
        require(self.stable <= through <= self.tail, 'stable prefix cannot retreat')
        self.stable = through

    def flush_page(self, page):
        require(page in self.pages, 'unknown page')
        # Same immutable page tuple supplies bytes and LSN; writes are atomic here.
        image = self.pages[page]
        if image[1] > self.stable:
            self.flush_log(image[1])
        self.disk[page] = image
        self.dirty.pop(page, None)

    def commit(self, tx):
        st = self.active(tx); r = self.append(tx, 'COMMIT')
        self.flush_log(r.lsn); st['status'] = 'COMMITTED'; st['marks'].clear()
        self.owner = {p: t for p, t in self.owner.items() if t != tx}
        return r

    def stable_records(self):
        return tuple(r for r in self.records if r.lsn <= self.stable)

    def values(self):
        return {p: image[0] for p, image in self.pages.items()}


@dataclass(frozen=True)
class Manifest:
    identity: tuple
    names: tuple
    begin: int
    redo: int
    retain: int
    end: int
    seed: tuple  # tx, firstLSN, lastLSN
    images: tuple  # page, value, pageLSN
    complete: bool


class Backup:
    def __init__(self, engine, identity=('db7', 'history1')):
        # One atomic metadata snapshot, serialized with registration in a real host.
        self.identity = tuple(identity)
        self.names = tuple(engine.pages)
        self.begin = engine.tail + 10
        self.redo = min([self.begin, *engine.dirty.values()])
        self.seed = tuple((t, st['first'], st['last']) for t, st in engine.tx.items()
                          if st['status'] in ('ACTIVE', 'ROLLING_BACK'))
        self.retain = min([self.redo, *(first for _, first, _ in self.seed)])
        self.images, self.copy_at = {}, {}
        self.end = None
        self.pages_stable = False
        self.complete = False

    def copy_page(self, engine, page):
        require(self.end is None and page in self.names and page not in self.images, 'copy state')
        self.images[page] = tuple(engine.disk[page])
        self.copy_at[page] = engine.tail

    def finish_copy(self, engine):
        require(self.end is None and set(self.images) == set(self.names), 'incomplete page set')
        self.end = engine.tail

    def stabilize_pages(self):
        require(self.end is not None, 'copy not complete')
        self.pages_stable = True

    def publish(self, archive):
        require(self.pages_stable and self.end is not None, 'pages not durable')
        archive_index(archive, self.retain, self.end)
        self.complete = True
        return self.manifest()

    def manifest(self):
        require(self.end is not None, 'copy not complete')
        return Manifest(self.identity, self.names, self.begin, self.redo, self.retain,
                        self.end, self.seed, tuple((p, *self.images[p]) for p in self.names), self.complete)


def archive_index(records, q, target):
    require(type(q) is int and q > 0 and q % 10 == 0, 'retention slot boundary')
    require(type(target) is int and target >= q-10 and target % 10 == 0, 'target slot boundary')
    idx = {}
    for r in records:
        require(isinstance(r, Record) and r.lsn not in idx, 'record type/duplicate LSN')
        idx[r.lsn] = r
    for lsn in range(q, target+10, 10):
        require(lsn in idx, 'archive hole at '+str(lsn))
    return idx


def restore(manifest, archive, target, identity=('db7', 'history1'), undo_limit=None):
    """Restore a trusted producer manifest; independently reject missing dependencies.
    undo_limit simulates a new crash after a bounded number of generated CLRs.
    Returned state is not a durable deployment or a ready service.
    """
    m = manifest
    require(m.complete and tuple(identity) == m.identity, 'manifest completion/identity')
    require(target >= m.end, 'target predates end of copy')
    require(m.retain <= m.redo <= m.begin and m.end >= m.begin-10, 'manifest bounds')
    require(len(set(m.names)) == len(m.names), 'duplicate manifest page')
    images = {p: (value, lsn) for p, value, lsn in m.images}
    require(len(images) == len(m.images) and set(images) == set(m.names), 'page manifest mismatch')
    require(all(0 <= lsn <= m.end and lsn % 10 == 0 for _, lsn in images.values()), 'future/torn page marker')
    idx = archive_index(archive, m.retain, target)
    e = Engine(m.names); e.pages = images.copy(); e.disk = images.copy()
    # Preserve the supplied global prefix where present; nothing is written to input.
    e.records = [idx[k] for k in range(m.retain, target+10, 10)]
    e.index = {r.lsn: r for r in e.records}; e.stable = target
    for tx, first, last in m.seed:
        require(tx not in e.tx and m.retain <= first <= last < m.begin, 'transaction seed')
        require(first in idx and last in idx and idx[first].tx == tx and idx[last].tx == tx, 'seed references')
        e.tx[tx] = {'status': 'ACTIVE', 'first': first, 'last': last, 'marks': []}
    redone, skipped = [], []
    for lsn in range(m.redo, target+10, 10):
        r = idx[lsn]
        if r.kind in ('UPDATE', 'CLR'):
            require(r.page in e.pages, 'log page outside backup domain')
            if e.pages[r.page][1] < r.lsn:
                e.apply(r); redone.append(r.lsn)
            else:
                skipped.append(r.lsn)
    for lsn in range(m.begin, target+10, 10):
        r = idx[lsn]
        if r.kind == 'BEGIN':
            require(r.tx not in e.tx, 'reused transaction identity')
            e.tx[r.tx] = {'status': 'ACTIVE', 'first': r.lsn, 'last': r.lsn, 'marks': []}
        else:
            require(r.tx in e.tx, 'transaction absent from seed and log')
            st = e.tx[r.tx]
            require(r.prev == st['last'], 'broken full transaction prev chain')
            st['last'] = r.lsn
            if r.kind == 'COMMIT': st['status'] = 'COMMITTED'
            elif r.kind == 'END': st['status'] = 'ENDED'
            else: require(r.kind in ('UPDATE', 'CLR'), 'unsupported log record')
    visits, generated = [], []
    before_undo = e.values()
    for tx, st in e.tx.items():
        if st['status'] != 'ACTIVE': continue
        cursor = st['last']
        while cursor:
            if undo_limit is not None and len(generated) >= undo_limit:
                return e, {'redone': redone, 'skipped': skipped, 'undo_visited': visits,
                           'new_clrs': generated, 'before_undo': before_undo, 'completed': False}
            require(cursor in e.index, 'missing undo history')
            visits.append(cursor)
            nxt, clr = e.undo_record(tx, cursor)
            require(nxt < cursor, 'undo chain does not descend')
            cursor = nxt
            if clr is not None: generated.append(clr.lsn)
        e.append(tx, 'END'); st['status'] = 'ENDED'
    return e, {'redone': redone, 'skipped': skipped, 'undo_visited': visits,
               'new_clrs': generated, 'before_undo': before_undo, 'completed': True}


def resume_restore(manifest, engine, identity=('db7', 'history1'), undo_limit=None):
    """Re-enter media recovery from the SAME durable backup after another crash.
    Keep its original images/TT seed and the archive from retain through stable,
    including every newly durable CLR/END. Discard tentative recovery pages and
    unstable log suffixes. No record before manifest.retain is requested.
    """
    return restore(manifest, engine.stable_records(), engine.stable,
                   identity=identity, undo_limit=undo_limit)


def restart(engine, undo_limit=None):
    """System-restart helper ONLY for a complete log beginning at slot 10.
    A truncated media-recovery history must use resume_restore and its backup.
    """
    require(engine.stable >= 10, 'test restart requires a nonempty stable log')
    require(10 in engine.index, 'restart requires log from 10; use resume_restore with original manifest')
    m = Manifest(('db7', 'history1'), tuple(engine.pages), 10, 10, 10, engine.stable,
                 (), tuple((p, *v) for p, v in engine.disk.items()), True)
    return restore(m, engine.stable_records(), engine.stable, undo_limit=undo_limit)


class Retention:
    """Each operation here represents one completed durable metadata operation.
    persist_progress stands for an independently completed consumer state write.
    """
    def __init__(self, end=160, segment=40):
        require(type(end) is int and end >= 0 and type(segment) is int and segment > 0, 'log domain')
        self.floor, self.end, self.segment = 0, end, segment
        self.active, self.generation, self.durable = {}, {}, {}

    def token(self, token):
        name, generation = token
        require(name in self.active and self.active[name][0] == generation, 'stale/unknown token')
        return name

    def register(self, name, start):
        require(name not in self.active, 'consumer already active')
        require(type(start) is int and self.floor <= start <= self.end, 'requested log no longer retained')
        g = self.generation.get(name, 0)+1
        self.generation[name] = g; self.active[name] = (g, start)
        self.durable[(name, g)] = start  # certified recoverable starting state
        return (name, g)

    def persist_progress(self, token, new):
        self.token(token)
        require(type(new) is int and self.durable[token] <= new <= self.end, 'progress must not retreat')
        self.durable[token] = new

    def advance(self, token, new):
        name = self.token(token); g, old = self.active[name]
        require(type(new) is int and old <= new <= self.durable[token], 'pin exceeds durable progress')
        self.active[name] = (g, new)

    def cancel(self, token):
        name = self.token(token)
        del self.active[name]
        del self.durable[token]
        # The per-name generation counter remains; an old token never revives.

    def finish(self, token):
        self.token(token)
        require(self.durable[token] == self.end, 'finite consumer not durably finished')
        self.cancel(token)

    def read(self, token, begin, end):
        name = self.token(token)
        require(self.active[name][1] <= begin <= end <= self.end and begin >= self.floor, 'read outside contract')
        return (begin, end)

    def collect(self):
        frontier = min([self.end, *(r for _, r in self.active.values())])
        stop = frontier//self.segment*self.segment
        deleted = [(x, x+self.segment) for x in range(self.floor, stop, self.segment)]
        # Persist floor before physical deletion. A crash may leak obsolete bytes.
        self.floor = stop
        return {'frontier': frontier, 'floor': self.floor, 'deleted': deleted}


def reject(call):
    try:
        call()
    except ValueError:
        return True
    raise AssertionError('invalid contract was accepted')


def demonstration():
    e = Engine(); e.begin('T'); e.write('T', 'A', 2); mark = e.savepoint('T')
    e.write('T', 'Q', 7); e.flush_page('Q')
    backup = Backup(e)
    require(e.dirty == {'A': 20} and e.disk['Q'] == (7, 30), 'begin snapshot has Q durable')
    backup.copy_page(e, 'A'); backup.copy_page(e, 'P'); backup.copy_page(e, 'Q')
    e.write('T', 'P', 9); partial = e.rollback_to('T', mark)
    require(e.tx['T']['status'] == 'ACTIVE' and e.owner['Q'] == 'T', 'partial rollback retains locks')
    e.write('T', 'R', 4); e.flush_page('R'); backup.copy_page(e, 'R'); backup.finish_copy(e)
    require(reject(lambda: backup.publish(e.stable_records())), 'publication requires stable copies')
    backup.stabilize_pages(); manifest = backup.publish(e.stable_records())
    crash70 = deepcopy(e)
    first, first_trace = restart(crash70, undo_limit=1)
    require(first_trace['new_clrs'] == [80], 'first recovery undo70')
    first.flush_log(first.tail)  # CLR durable, compensation page deliberately not flushed.
    second, second_trace = restart(first)
    require(second.values() == {p: 0 for p in e.pages}, 'second restart clears transaction')
    e.commit('T'); r80, trace80 = restore(manifest, e.stable_records(), 80)
    e.begin('U'); e.write('U', 'Q', 8); e.begin('V'); e.write('V', 'P', 6); e.commit('V')
    r130, trace130 = restore(manifest, e.stable_records(), 130)
    require(r80.values() == {'A':2,'P':0,'Q':0,'R':4}, 'target80')
    require(r130.values() == {'A':2,'P':6,'Q':0,'R':4}, 'target130')
    failures = []
    for tag, fn in [
        ('early_target',lambda:restore(manifest,e.stable_records(),60)),
        ('missing_CLR60',lambda:restore(manifest,tuple(r for r in e.stable_records() if r.lsn!=60),130)),
        ('missing_UPDATE120',lambda:restore(manifest,tuple(r for r in e.stable_records() if r.lsn!=120),130)),
        ('wrong_history',lambda:restore(manifest,e.stable_records(),130,('db7','history2'))),
    ]:
        reject(fn);failures.append(tag)
    # A genuinely truncated media-recovery archive: the committed prefix is gone.
    cut = Engine(('A', 'B')); cut.begin('OLD'); cut.write('OLD', 'A', 5)
    cut.commit('OLD'); cut.flush_page('A'); cut.begin('NEW'); cut.write('NEW', 'B', 9)
    cb = Backup(cut)
    for p in cut.pages: cb.copy_page(cut, p)
    cb.finish_copy(cut); cb.stabilize_pages(); cut.flush_log(cut.tail)
    cm = cb.publish(tuple(r for r in cut.stable_records() if r.lsn >= cb.retain))
    once, once_trace = restore(cm, tuple(r for r in cut.stable_records() if r.lsn >= cm.retain),
                               50, undo_limit=1)
    require((cm.retain, cm.redo, cm.begin, cm.end) == (40, 50, 60, 50), 'truncated archive bounds')
    once.flush_log(once.tail)  # durable CLR60; tentative page B=0 need not survive.
    reject(lambda: restart(once))
    again, again_trace = resume_restore(cm, once)
    require([r.lsn for r in once.records] == [40, 50, 60], 'no silently retained early prefix')
    require(again.values() == {'A': 5, 'B': 0} and again_trace['new_clrs'] == [],
            'media reentry preserves old commit and skips already compensated update')
    g = Retention(); redo=g.register('redo',90); undo=g.register('undo',10)
    back=g.register('backup',10); replica=g.register('replica',70)
    stages=[g.collect()];g.cancel(undo);g.cancel(back);stages.append(g.collect())
    reject(lambda:g.advance(replica,100))
    g.persist_progress(replica,100);g.advance(replica,100);stages.append(g.collect())
    reject(lambda:g.register('late',50));late=g.register('late',80)
    g.cancel(late);new=g.register('late',80)
    reject(lambda:g.advance(late,90));reject(lambda:g.finish(late));reject(lambda:g.read(late,80,90))
    require(new != late, 'new generation')
    return {'manifest':manifest.__dict__,'copy_log_times':backup.copy_at,'partial':partial,
            'restore80':{'values':r80.values(),**trace80},'restore130':{'values':r130.values(),**trace130},
            'crash70_first':first_trace,'crash70_second':second_trace,'second_values':second.values(),
            'truncated_media_reentry':{'manifest':cm.__dict__, 'first':once_trace,
                'stable_archive':[r.lsn for r in once.stable_records()],
                'second':again_trace, 'values':again.values()},
            'rejected':failures,'retention_stages':stages,'stale_token':late,'new_token':new,
            'unsafe_advance_counterexample':{'durable_restart':70,'falsely_claimed_pin':100,'deleted_until':80}}


def checks():
    rng=random.Random(131009);counts={'savepoint_operations':0,'restarts':0,'fuzzy_images':0,'retention_events':0}
    # A literal value snapshot oracle, independent of log cursors and CLR chains.
    for run in range(160):
        e=Engine();e.begin('T');oracle={p:0 for p in e.pages};marks=[]
        for _ in range(55):
            op=rng.randrange(5)
            if op<2:
                p=rng.choice(list(oracle));v=rng.randrange(-9,10)
                e.write('T',p,v);oracle[p]=v
            elif op==2 or not marks:
                t=e.savepoint('T');marks.append((t,oracle.copy()))
            elif op==3:
                i=rng.randrange(len(marks));token,snap=marks[i]
                e.rollback_to('T',token);oracle=snap.copy();marks=marks[:i+1]
            else:
                i=rng.randrange(len(marks));e.release('T',marks[i][0]);marks=marks[:i]
            require(e.values()==oracle, 'snapshot rollback oracle')
            if rng.randrange(5)==0:e.flush_page(rng.choice(list(oracle)))
            counts['savepoint_operations']+=1
        e.flush_log(e.tail)
        for mask in range(16):
            disk=deepcopy(e)
            for j,p in enumerate(oracle):
                if mask>>j&1:disk.flush_page(p)
            restored,_=restart(disk)
            require(not any(restored.values().values()), 'loser full rollback')
            counts['restarts']+=1
        # All accepted fuzzy images are independent per-page prefixes of a complete log.
        e.commit('T');hist={p:[(0,0)]for p in oracle}
        for rec in e.records:
            if rec.kind in ('UPDATE','CLR'):hist[rec.page].append((rec.after,rec.lsn))
        for _ in range(12):
            images=tuple((p,*rng.choice(hist[p]))for p in oracle)
            end=max(x[2]for x in images)
            m=Manifest(('db7','history1'),tuple(oracle),10,10,10,end,(),images,True)
            got,_=restore(m,e.stable_records(),e.stable)
            require(got.values()==oracle, 'fuzzy images plus commit')
            counts['fuzzy_images']+=1
    # Independent needed-position union, compared after every random event.
    for run in range(250):
        g=Retention(end=400,segment=20);tokens={};needs={}
        for step in range(100):
            name='c'+str(rng.randrange(7));op=rng.randrange(4)
            if name not in tokens:
                start=rng.randrange(g.floor,g.end+1);tokens[name]=g.register(name,start);needs[name]=start
            elif op==0:
                t=tokens.pop(name);g.cancel(t);needs.pop(name)
            elif op==1:
                t=tokens[name];new=rng.randrange(needs[name],g.end+1)
                g.persist_progress(t,new);g.advance(t,new);needs[name]=new
            else:
                needed=set()
                for start in needs.values():needed.update(range(start,g.end))
                before=g.floor;row=g.collect()
                deleted=set(range(before,g.floor))
                require(not(deleted&needed),'deleted a consumer needed position')
                expected=min([g.end,*needs.values()])//g.segment*g.segment
                require(g.floor==expected and row['frontier']==min([g.end,*needs.values()]),'exact segment frontier')
            for name,t in tokens.items():require(g.floor<=needs[name]<=g.durable[t],'recoverable consumer')
            counts['retention_events']+=1
    schedules=[]
    for order in permutations('DPG'):
        durable,pin,floor=70,70,40;lost=False
        for event in order:
            if event=='D':durable=100
            elif event=='P':pin=100  # deliberately unsafe variant bypassing advance guard
            else:floor=min(90,pin)//40*40
            lost |= floor>durable
        schedules.append({'order':''.join(order),'lost_before_next_event':lost})
    require([x['order']for x in schedules if x['lost_before_next_event']]==['PGD'],'wrong-order loss witness')
    return {**counts,'three_event_orders':schedules}


if __name__=='__main__':
    print(json.dumps({'status':'PASS','demonstration':demonstration(),'checks':checks()},ensure_ascii=False,indent=2))