Two-Phase Commit — সবাই একসাথে, নয়তো কেউই না

"""two_phase_commit.py — coordinator + N participants, hand-traceable, deterministic"""

class Participant:
    def __init__(self, name, vote):
        self.name = name
        self.vote = vote              # pre-decided for this sim: 'YES' or 'NO'
        self.state = 'INIT'
        self.locked = False

    def prepare(self):
        if self.vote == 'YES':
            self.locked = True
            self.state = 'IN-DOUBT'   # committed to whatever coordinator decides
        else:
            self.state = 'ABORTED'    # NO-voter can abort unilaterally, right now
        return self.vote

    def receive_decision(self, decision):
        self.state = decision
        self.locked = False

    def status(self):
        return f"{self.name}:{self.state}{' [locked]' if self.locked else ''}"


class Coordinator:
    def __init__(self, participants):
        self.participants = participants
        self.log = None               # durable decision log; None = undecided

    def prepare_phase(self):
        print("  PREPARE ->", ', '.join(p.name for p in self.participants))
        votes = {}
        for p in self.participants:
            v = p.prepare()
            votes[p.name] = v
            print(f"    {p.name} votes {v:<3} -> {p.status()}")
        return votes

    def decide(self, votes):
        decision = 'COMMIT' if all(v == 'YES' for v in votes.values()) else 'ABORT'
        print(f"  decision computed: {decision}")
        return decision

    def log_decision(self, decision):
        self.log = decision           # fsync BEFORE telling anyone — the whole trick
        print(f"  decision durably logged: {self.log}")

    def broadcast(self, decision):
        for p in self.participants:
            p.receive_decision('COMMITTED' if decision == 'COMMIT' else 'ABORTED')
            print(f"    {p.name} <- {decision:<6} -> {p.status()}")

    def recover(self):
        print("  RECOVERY: fresh coordinator process reads the durable decision log")
        if self.log is None:
            print("    log empty -> crashed before deciding -> safe default ABORT")
            decision = 'ABORT'
        else:
            print(f"    log has decision={self.log} -> rebroadcast, no new decision needed")
            decision = self.log
        self.broadcast(decision)
        return decision


print("=== Case A: happy path — সব participant YES ===")
psA = [Participant('OrderDB', 'YES'), Participant('InventoryDB', 'YES'), Participant('PaymentDB', 'YES')]
cA = Coordinator(psA)
votesA = cA.prepare_phase()
dA = cA.decide(votesA)
cA.log_decision(dA)
cA.broadcast(dA)

print("\n=== Case B: PaymentDB vote দেয় NO ===")
psB = [Participant('OrderDB', 'YES'), Participant('InventoryDB', 'YES'), Participant('PaymentDB', 'NO')]
cB = Coordinator(psB)
votesB = cB.prepare_phase()
dB = cB.decide(votesB)
cB.log_decision(dB)
cB.broadcast(dB)

print("\n=== Case C: coordinator log-এর পরে, broadcast-এর আগে crash করে ===")
psC = [Participant('OrderDB', 'YES'), Participant('InventoryDB', 'YES'), Participant('PaymentDB', 'YES')]
cC = Coordinator(psC)
votesC = cC.prepare_phase()
dC = cC.decide(votesC)
cC.log_decision(dC)
print("  !! coordinator process dies এখানে — logged, কিন্তু broadcast হয়নি !!")
for p in psC:
    print(f"    still blocked: {p.status()}")
print("  --- সময় গেল; নতুন coordinator process ওঠে, একই durable log খোলে ---")
cC.recover()
assert all(p.state == 'COMMITTED' and not p.locked for p in psC)
print("  PASS — recovery-ই একমাত্র মুক্তি, আর ঠিক ওই decision-ই ফিরল যা log-এ ছিল")

প্রত্যাশিত output (deterministic — হুবহু এটাই আসবে):

=== Case A: happy path — সব participant YES ===
  PREPARE -> OrderDB, InventoryDB, PaymentDB
    OrderDB votes YES -> OrderDB:IN-DOUBT [locked]
    InventoryDB votes YES -> InventoryDB:IN-DOUBT [locked]
    PaymentDB votes YES -> PaymentDB:IN-DOUBT [locked]
  decision computed: COMMIT
  decision durably logged: COMMIT
    OrderDB <- COMMIT -> OrderDB:COMMITTED
    InventoryDB <- COMMIT -> InventoryDB:COMMITTED
    PaymentDB <- COMMIT -> PaymentDB:COMMITTED

=== Case B: PaymentDB vote দেয় NO ===
  PREPARE -> OrderDB, InventoryDB, PaymentDB
    OrderDB votes YES -> OrderDB:IN-DOUBT [locked]
    InventoryDB votes YES -> InventoryDB:IN-DOUBT [locked]
    PaymentDB votes NO  -> PaymentDB:ABORTED
  decision computed: ABORT
  decision durably logged: ABORT
    OrderDB <- ABORT  -> OrderDB:ABORTED
    InventoryDB <- ABORT  -> InventoryDB:ABORTED
    PaymentDB <- ABORT  -> PaymentDB:ABORTED

=== Case C: coordinator log-এর পরে, broadcast-এর আগে crash করে ===
  PREPARE -> OrderDB, InventoryDB, PaymentDB
    OrderDB votes YES -> OrderDB:IN-DOUBT [locked]
    InventoryDB votes YES -> InventoryDB:IN-DOUBT [locked]
    PaymentDB votes YES -> PaymentDB:IN-DOUBT [locked]
  decision computed: COMMIT
  decision durably logged: COMMIT
  !! coordinator process dies এখানে — logged, কিন্তু broadcast হয়নি !!
    still blocked: OrderDB:IN-DOUBT [locked]
    still blocked: InventoryDB:IN-DOUBT [locked]
    still blocked: PaymentDB:IN-DOUBT [locked]
  --- সময় গেল; নতুন coordinator process ওঠে, একই durable log খোলে ---
  RECOVERY: fresh coordinator process reads the durable decision log
    log has decision=COMMIT -> rebroadcast, no new decision needed
    OrderDB <- COMMIT -> OrderDB:COMMITTED
    InventoryDB <- COMMIT -> InventoryDB:COMMITTED
    PaymentDB <- COMMIT -> PaymentDB:COMMITTED
  PASS — recovery-ই একমাত্র মুক্তি, আর ঠিক ওই decision-ই ফিরল যা log-এ ছিল

তিনটা পর্যবেক্ষণ: (১) Case A-তে vote দেওয়ার সাথে সাথেই সবাই [locked] হয়ে গেল, commit না হওয়া পর্যন্ত সেটা যায়নি — lock-window খালি চোখে; (২) Case B-তে PaymentDB নিজে থেকেই ABORTED (কখনো lock নেয়নি), কিন্তু OrderDB/InventoryDB আগে থেকেই lock নিয়ে বসেছিল — তাদের lock ছাড়তে পুরো broadcast round লাগল, blocking-এর দাম এক NO-vote-এও এড়ানো যায় না; (৩) Case C-তে still blocked লাইনগুলোই blocking-এর সংজ্ঞা, আর RECOVERY অংশটাই দেখায় কেন log-write-ই commit-এর আসল মুহূর্ত।