Saga ও Distributed Transactions — 2PC-র বিকল্প পথ

লক্ষ্য: crash mid-compensation হলেও restart-এ saga শুদ্ধ শেষ হবে — প্রতিটা effect ঠিক একবার।

"""saga_orchestrator.py — journal-based resumable saga engine"""
import json, os, tempfile
from dataclasses import dataclass
from typing import Callable

@dataclass
class Step:
    name: str
    action: Callable[[dict], str]
    compensate: Callable[[dict], str]
    retries: int = 2

class SagaCrashed(Exception):
    pass

class Orchestrator:
    def __init__(self, steps, world, journal_path, fail_at=None):
        self.steps, self.world, self.journal_path = steps, world, journal_path
        self.fail_at, self.effects = fail_at, {}   # effect-key -> count

    def _journal(self, entry):
        with open(self.journal_path, 'a') as fh:
            fh.write(json.dumps(entry) + '\n')

    def _recover(self):
        """restart-এ journal পড়ে completed-effect map পুনর্নির্মাণ"""
        try:
            with open(self.journal_path) as fh:
                for line in fh:
                    e = json.loads(line)
                    if e.get('s') == 'done':
                        self.effects[e['k']] = 1
        except FileNotFoundError:
            pass

    def _maybe_crash(self, phase, idx):
        if self.fail_at == (phase, idx):
            raise SagaCrashed(f'crash @ {phase}{idx}')

    def _run_effect(self, key, fn, attempts):
        if self.effects.get(key):            # exactly-once guard (dedup)
            return
        self._journal({'k': key, 's': 'intent'})
        last = None
        for _ in range(attempts):            # retry policy
            try:
                result = fn(self.world)
                self.effects[key] = 1
                self._journal({'k': key, 's': 'done', 'r': result})
                return result
            except Exception as err:
                last = err
        raise RuntimeError(f'{key} failed: {last}')

    def run(self, i=1, phase='forward'):
        self._recover()
        rng = range(i, len(self.steps) + 1) if phase == 'forward' \
            else range(i, 0, -1)
        for n in rng:
            st = self.steps[n - 1]
            if phase == 'forward':
                try:
                    self._run_effect(f'T{n}', st.action, st.retries)
                except RuntimeError as err:      # business-fail → rollback
                    print(f'  >> {err}; cascade শুরু C{n-1}')
                    return self.run(n - 1, 'compensate')
                self._maybe_crash(phase, n)      # crash = process-death
            else:
                self._maybe_crash(phase, n)      # mid-compensation crash!
                self._run_effect(f'C{n}', st.compensate, st.retries)
        rolled = any(k.startswith('C') for k in self.effects)
        return 'ROLLED-BACK' if rolled else 'COMPLETE'

# ── demo domain: world = shared fake-state dict ──
def mk(tag, ok, undo):
    return Step(tag,
                lambda w: (w.setdefault(tag, []).append(ok), ok)[1],
                lambda w: (w.setdefault(tag, []).append(undo), undo)[1])

steps = [
    mk('order',  'PENDING',  'CANCELLED'),
    mk('flight', 'HELD',     'RELEASED'),
    mk('hotel',  'RESERVED', 'ROOMLESS'),
]
boom = lambda w: (_ for _ in ()).throw(RuntimeError('no rooms'))
steps[2] = Step('hotel', boom, lambda w: 'room released')

tmp = os.path.join(tempfile.mkdtemp(), 'journal.ndjson')
print('=== run 1: crash mid-compensation (C1-এর ঠিক আগে process-death) ===')
orc = Orchestrator(steps, {}, tmp, fail_at=('compensate', 1))
try:
    orc.run()
except SagaCrashed as err:
    print(f'  died: {err}; journaled-done: {sorted(orc.effects)}')

print('=== run 2: restart — একই journal ফাইল, নতুন process ===')
orc2 = Orchestrator(steps, {}, tmp)
verdict = orc2.run()
print(f'  verdict={verdict}')
print(f'  effects={orc2.effects}')
assert all(v == 1 for v in orc2.effects.values()), 'exactly-once ভাঙল!'
assert orc2.effects.get('C2') == 1   # C2 আবার চলেনি — dedup-guard কাজ করেছে
print('  PASS — crash-পরেও প্রতিটা effect ঠিক একবার')

Journal-এর নমুনা content:

{"k": "T1", "s": "done", "r": "PENDING"}
{"k": "T2", "s": "done", "r": "HELD"}
{"k": "C2", "s": "intent"}
{"k": "C2", "s": "done", "r": "RELEASED"}   ← এখানেই process মরেছিল

তিনটা পাঠ:

• Exactly-once এখানেও illusion — _run_effect বাস্তবে at-least-once চালায়, effects-guard duplicate-effect আটকায়; end-to-end সত্য।
• Intent-এর পরে done-entry — restart শুধু done replay করে; half-done state বলে কিছু নেই। Temporal/Cadence এই journal-কেই event-history বলে disk+replication-এ রাখে — মূলনীতি এক।

নিজে বাড়ান:

1. Compensate-phase retry-exhaustion handle করুন — human- escalation flag জমা দিন (poison-saga queue)
2. দুই concurrent saga এক world-এ চালিয়ে lost update দেখুন; version-check যোগ করে আটকান
3. Journal-টাই append-only event-log বানান — event sourcing- এর প্রিভিউ!