WAL ও ARIES Recovery — crash-এর পরে হিসাব

লক্ষ্য: example section-এর হাতে-trace-টা স্বয়ংক্রিয় — crash যেকোনো মুহূর্তে হলেও ফল এক থাকে তা প্রমাণসহ।

import random
from dataclasses import dataclass

NIL = -1

@dataclass
class Rec:
    lsn: int
    kind: str                    # 'begin'|'update'|'commit'|'clr'
    txid: int
    page: str = ''
    old: object = None
    new: object = None
    undo_next: int = NIL         # শুধু CLR ব্যবহার করে
class Node:
    """WAL + buffer pool + disk — steal/no-force, ARIES-lite recovery"""
    def __init__(self):
        self.log = []                         # stable storage — crash-proof
        self.pool = {}                        # volatile buffer pool
        self.pool_lsn = {}                    # volatile
        self.disk = {}                        # survives
        self.disk_lsn = {}                    # page header-এর page_LSN
        self.holds = {}                       # strict-2PL অভিনয়: page -> txid

    def append(self, **kw) -> int:
        rec = Rec(len(self.log), **kw)        # list append == instant durable
        self.log.append(rec)
        return rec.lsn

    def current(self, page):
        return self.pool.get(page, self.disk.get(page))

    def begin(self, tx):
        self.append(kind='begin', txid=tx)

    def write(self, tx, page, new):
        if self.holds.get(page, tx) != tx:
            raise RuntimeError("strict-2PL: page অন্য active txn-এর হাতে")
        self.holds[page] = tx
        old = self.current(page)
        lsn = self.append(kind='update', txid=tx, page=page, old=old, new=new)
        self.pool[page] = new                 # WAL-first: log-এর *পরে* dirty
        self.pool_lsn[page] = lsn

    def commit(self, tx):
        self.append(kind='commit', txid=tx)   # এই LSN durable == commit point
        for p in [p for p, o in self.holds.items() if o == tx]:
            del self.holds[p]

    def steal_somewhere(self):
        """buffer manager-এর স্বাধীনতা: যেকোনো dirty page disk-এ"""
        if not self.pool:
            return
        page = random.choice(list(self.pool))
        self.disk[page] = self.pool.pop(page)     # WAL rule: log already durable
        self.disk_lsn[page] = self.pool_lsn.pop(page)

    def crash(self):
        self.pool.clear()                     # সব volatile state শেষ
        self.pool_lsn.clear()

    def recover(self):
        # PASS 1: analysis — losers বের করা
        losers = set()
        for r in self.log:
            if r.kind == 'begin':
                losers.add(r.txid)
            elif r.kind == 'commit':
                losers.discard(r.txid)
        # PASS 2: redo — history repeat (uncommitted-সহ), idempotent শর্তে
        for r in self.log:
            if r.kind not in ('update', 'clr'):
                continue
            if self.disk_lsn.get(r.page, NIL) < r.lsn:
                self.disk[r.page] = r.new
                self.disk_lsn[r.page] = r.lsn
        # PASS 3: undo — backward, losers-এর জন্য CLR emission
        pending = {}
        for i, r in enumerate(self.log):      # প্রতিটা loser-এর শেষ update
            if r.kind == 'update' and r.txid in losers:
                pending[r.txid] = i
        for i in range(len(self.log) - 1, -1, -1):
            r = self.log[i]
            if r.kind == 'update' and pending.get(r.txid) == i:
                self.disk[r.page] = r.old
                self.disk_lsn[r.page] = self.append(
                    kind='clr', txid=r.txid, page=r.page,
                    new=r.old, undo_next=pending[r.txid])
                prev = NIL
                for j in range(i - 1, -1, -1):
                    if self.log[j].kind == 'update' and \
                            self.log[j].txid == r.txid:
                        prev = j
                        break
                pending[r.txid] = prev


def committed_prefix(work):
    """স্বাধীন যাচাই: committed-prefix অনুযায়ী প্রত্যাশিত অবস্থা"""
    expect, staged, holds = {}, {}, {}
    for op in work:
        if op[0] == 'w':
            _, tx, page, val = op
            if holds.get(page, tx) == tx:
                holds[page] = tx
                staged.setdefault(tx, {})[page] = val
        elif op[0] == 'c':
            expect.update(staged.get(op[1], {}))
            holds = {p: o for p, o in holds.items() if o != op[1]}
    return expect


def trial(seed):
    rng = random.Random(seed)
    node, work, active, tx = Node(), [], [], 0
    for step in range(rng.randint(15, 45)):
        act = rng.random()
        if act < 0.25 or not active:
            tx += 1
            node.begin(tx)
            active.append(tx)
            work.append(('b', tx))
        elif act < 0.8:
            t = rng.choice(active)
            page, val = f"P{rng.randint(1, 4)}", rng.randint(0, 99)
            try:
                node.write(t, page, val)
                work.append(('w', t, page, val))
            except RuntimeError:
                pass                          # lock-conflict — বাদ
        else:
            t = rng.choice(active)
            active.remove(t)
            node.commit(t)
            work.append(('c', t))        if rng.random() < 0.3:
            node.steal_somewhere()
    node.crash()
    node.recover()
    once = dict(node.disk)
    node.recover()                            # idempotence: দ্বিতীয়বার
    assert dict(node.disk) == once, "recovery idempotent নয়!"
    # শুধু loser-রা ছোঁয়া pages None-এ ফেরে — prefix-এ ওরা নেই
    final = {p: v for p, v in once.items() if v is not None}
    assert final == committed_prefix(work), f"{final} != প্রত্যাশিত"
    return True


for s in range(300):
    trial(s)
print("300 crash-point সবই committed-prefix-এ; recovery idempotent।")

নিজে বাড়ান:

1. Checkpoint — প্রতি N record-এ dirty-list + active-set snapshot; redo সেখান থেকে
2. steal_somewhere adversarial — recent-est page flush; ফল একই দেখান