Replication — streaming আর logical-এর দুই পথ

লক্ষ্য: replication-lag-এর পুরো জীবনচক্র simulation — স্বাভাবিক ঢেউ, partition-এর প্রশস্ত উপত্যকা, heal-এর পরে খাড়া পুনরুদ্ধার।

import random

APPLY_RATE = 8          # standby প্রতি tick-এ সর্বোচ্চ এত frame apply

def simulate(delay_max=2, partition=(40, 71), horizon=140, seed=11):
    rng = random.Random(seed)
    lsn = applied = 0
    wire = []                       # (arrive_tick, frame)
    held = []                       # partition-এ আটকে থাকা frames
    lag = {}
    phase = {}
    healed_at = None
    caught_up_tick = None

    for t in range(1, horizon + 1):
        in_partition = partition[0] <= t <= partition[1]

        # --- primary: ~3 frames/tick emit ---
        for _ in range(rng.randint(2, 4)):
            lsn += 1
            if in_partition:
                held.append(lsn)             # WAL retained — হারায় না
            else:
                wire.append((t + rng.randint(1, delay_max), lsn))

        # --- heal: retained frames এখন পথে ---
        if healed_at is None and t == partition[1] + 1:
            healed_at = t
            for f in held:
                wire.append((t + rng.randint(1, delay_max), f))
            held.clear()

        # --- standby: পৌঁছানো frames, capped-rate apply ---
        arrived = sorted(fr for at, fr in wire if at <= t)
        wire = [(at, fr) for at, fr in wire if at > t]
        for fr in arrived[:APPLY_RATE]:
            applied = max(applied, fr)
        # cap ছাড়ানো frames হারায় না — পরের tick-এর queue-তে ফেরে
        wire.extend((t + 1, fr) for fr in arrived[APPLY_RATE:])
        lag[t] = lsn - applied
        if healed_at and caught_up_tick is None and lag[t] == 0:
            caught_up_tick = t
        phase[t] = ("PARTITIONED" if in_partition else
                    "CATCHING UP" if healed_at and lag[t] else "normal")

    return lag, phase, healed_at, caught_up_tick


lag, phase, healed, done = simulate()
print("tick  lag  অবস্থা")
for t in range(10, len(lag) + 1, 10):
    print(f"{t:>4} {lag[t]:>4}  {phase[t]}")

worst_t = max(lag, key=lag.get)
divergence = sum(1 for v in lag.values() if v > 0)
print(f"\nmax lag            : {max(lag.values())} frames "
      f"(tick {worst_t})")
print(f"divergence window  : {divergence} ticks")
print(f"heal -> caught up  : tick {healed} -> {done} "
      f"({done - healed} ticks)")

প্রত্যাশিত output-এর আকার:

tick  lag  অবস্থা
  10    2  normal
  20    3  normal
  30    2  normal
  40   14  PARTITIONED
  50   74  PARTITIONED
  60  134  PARTITIONED
  70  194  PARTITIONED
  80  120  CATCHING UP
  90   40  CATCHING UP
 100    0  normal
 ...

max lag            : 196 frames (tick 71)
divergence window  : 62 ticks
heal -> caught up  : tick 71 -> 97 (26 ticks)

তিনটা পাঠ:

• Partition-এ lag প্রায় linearly বাড়ে — write-rate × সময়, এটাই async replication-এর RPO-র গাণিতিক ভিত্তি
• Heal-এর পরে সোজা শূন্যে না — APPLY_RATE-cap-এর কারণে catch-up-ও সময় নেয়; বাস্তব standby-র replay I/O-ও এমনি capped
• held frames হারায়নি — WAL-retention-এর মডেল; কিন্তু simulation-এ disk-unlimited, বাস্তবে সেটাই slot-এর বিল

নিজে বাড়ান:

1. held-এর জন্য disk-budget যোগ করুন — overflow হলে frames drop (slot-invalid) — আর দেখুন divergence permanent হয়ে যায়
2. Sync-mode অনুকরণ — client-ack শুধু তখনই যখন applied >= lsn; partition-এ availability কী হয়, গুনুন (CAP-এর প্রিভিউ!)
3. Apply-rate-কে lag-dependent বানান — বড় backlog মানে তুলনায় দ্রুত drain (বাস্তব restore_command/walreceiver-এর আচরণ)