Gossip Protocol ও CRDT — ছড়িয়ে পড়া আর conflict-free গণিত

"""crdts.py — four CRDTs + property-based fuzz (~120 line)"""
import random
from dataclasses import dataclass, field


@dataclass
class GCounter:
    slots: dict = field(default_factory=dict)   # node -> count

    def inc(self, node, by=1):
        self.slots[node] = self.slots.get(node, 0) + by   # নিজের slot-ই

    def value(self):
        return sum(self.slots.values())

    def merge(self, other):
        for node, c in other.slots.items():
            self.slots[node] = max(self.slots.get(node, 0), c)


@dataclass
class PNCounter:
    pos: GCounter = field(default_factory=GCounter)
    neg: GCounter = field(default_factory=GCounter)

    def inc(self, node, by=1):
        (self.pos if by >= 0 else self.neg).inc(node, abs(by))

    def value(self):
        return self.pos.value() - self.neg.value()

    def merge(self, other):
        self.pos.merge(other.pos)
        self.neg.merge(other.neg)


@dataclass
class LWWRegister:
    value=None; ts=0; writer=''
    def set(self, value, ts, writer):   # strict > — equal-ts-এ পুরনো জেতে
        if (ts, writer) > (self.ts, self.writer):
            self.value, self.ts, self.writer = value, ts, writer
    def merge(self, other):
        self.set(other.value, other.ts, other.writer)


@dataclass
class ORSet:
    adds: dict = field(default_factory=dict)    # tag -> elem
    rems: set = field(default_factory=set)      # tombstoned tags

    def add(self, elem, tag):
        self.adds[tag] = elem

    def remove(self, elem):
        self.rems |= {t for t, e in self.adds.items() if e == elem}

    def elements(self):
        return {e for t, e in self.adds.items() if t not in self.rems}

    def merge(self, other):
        merged = dict(self.adds); merged.update(other.adds)
        self.adds = merged
        self.rems |= other.rems


def random_schedule(rng, ops_per_replica=12):
    nodes = ['n1', 'n2', 'n3']
    g = {n: GCounter() for n in nodes}
    pn = {n: PNCounter() for n in nodes}
    lw = {n: LWWRegister() for n in nodes}
    os_ = {n: ORSet() for n in nodes}
    for i in range(ops_per_replica):
        n = rng.choice(nodes)
        g[n].inc(n)
        pn[n].inc(n, rng.choice([1, 1, -1]))
        lw[n].set(f'v{i}', rng.randrange(100), n)
        if rng.random() < 0.6:
            os_[n].add(f'e{rng.randrange(4)}', f'tag{n}-{i}')
        else:
            os_[n].remove(f'e{rng.randrange(4)}')

    # random sync-graph: এলোমেলো pair-merge — order ইচ্ছামতোই থাক
    for _ in range(40):
        a, b = rng.sample(nodes, 2)
        g[a].merge(g[b]);   pn[a].merge(pn[b])
        lw[a].merge(lw[b]); os_[a].merge(os_[b])
        g[b].merge(g[a]);   pn[b].merge(pn[a])
        lw[b].merge(lw[a]); os_[b].merge(os_[a])

    assert len({g[n].value() for n in nodes}) == 1, 'G diverged!'
    assert len({pn[n].value() for n in nodes}) == 1, 'PN diverged!'
    assert len({(lw[n].value, lw[n].ts, lw[n].writer)
                for n in nodes}) == 1, 'LWW diverged!'
    assert len({frozenset(os_[n].elements())
                for n in nodes}) == 1, 'ORSet diverged!'


def intent_cases():                    # documented intent-preservation cases
    a, b = GCounter(), GCounter()      # duplicate delivery harmless?
    a.inc('n1'); a.inc('n1'); b.merge(a); b.merge(a)
    assert b.value() == 2, 'idempotent merge broken'

    p = PNCounter(); p.inc('n1', 5); p.inc('n2', -3)
    assert p.value() == 2, 'PN arithmetic broken'

    r1, r2 = LWWRegister(), LWWRegister()
    r1.set('late', 10, 'n1'); r2.set('early', 5, 'n2')
    r1.merge(r2)
    assert r1.value == 'late', 'LWW ordering broken'

    s1, s2 = ORSet(), ORSet()          # add-wins under concurrency?
    s1.add('x', 't1'); s2.remove('x')
    s1.merge(s2); s2.merge(s1)
    assert 'x' in s1.elements() and 'x' in s2.elements(), 'add-wins broken'
    print('intent cases: PASS (idempotency, PN-arithmetic, LWW-order, add-wins)')


rng = random.Random(42)
for i in range(300):                        # property-based fuzz
    random_schedule(rng)
print('fuzz: 300 random schedules — convergence holds everywhere')
intent_cases()

নমুনা output:

fuzz: 300 random schedules — convergence holds everywhere
intent cases: PASS (idempotency, PN-arithmetic, LWW-order, add-wins)

তিনটা পাঠ:

• Fuzz specific schedule না, property test করে: যেকোনো interleaving-এ convergence — ৩০০ schedule মানে ৩০০ বিশ্ব, সবখানে একই শেষ। Intent-case আরও কঠিন: ‘মিলেছে’ না, ঠিকটা মিলেছে কি। LWW-র strict >-এ equal-ts- এ পুরনো জেতে — production-এ incarnation-number দিয়ে refine হয়।

নিজে বাড়ান:

1. MV-Register — concurrent write দুটোই রেখে client-কে resolve করতে দেয়
2. Delta-mode — last-sync watermark রেখে শুধু delta; bandwidth-সাশ্রয় মাপুন
3. Fuzz-এ fault-model — half-merged sync drop-পরবর্তী recovery converge করে?