Vector Clock Library

কেন এই প্রজেক্ট

Lamport clock শেখার পর অনেকের মনে থাকে একটাই লাইন: “timestamp ছোট মানে আগে ঘটেছে।” কিন্তু ওই এক লাইনে একটা ফাঁদ লুকিয়ে আছে — Lamport clock-এ L(a) < L(b) দেখলে বলা যায় না a আসলেই আগে ঘটেছে। Concurrency ধরার যন্ত্রটা আলাদা, আর সেটাই vector clock — Dynamo, Riak, Cassandra-র ভেতরে conflict detection-এর মূল পার্টস।

এই প্রজেক্টের আসল শিক্ষা একটাই: compare কোনো boolean নয়, চার-মুখো partial order। বেশিরভাগ bug তাইও ওখানেই জন্মায় — মানুষ == আর < দিয়ে clock তুলনা করতে গিয়ে CONCURRENT case হারিয়ে ফেলে, আর সেই রাতে দুই replica-এ দুই রকম shopping cart। লাইব্রেরিটা নিজের হাতে বানালে happens-before লেসনের সংজ্ঞাগুলো মুখস্থ থেকে হাতের টুলে বদলে যাবে।

ধাপে ধাপে

১. VectorClock core — চার-mুখো compare

from enum import Enum
from typing import Dict

NodeId = str
VC = Dict[NodeId, int]

class Order(Enum):
    EQUAL      = "equal"       # একই জ্ঞান
    BEFORE     = "before"      # a happens-before b
    AFTER      = "after"       # b happens-before a
    CONCURRENT = "concurrent"  # কেউ কারও পূর্বসূরী নয়

def compare(a: VC, b: VC) -> Order:
    keys = set(a) | set(b)
    le = all(a.get(k, 0) <= b.get(k, 0) for k in keys)
    ge = all(a.get(k, 0) >= b.get(k, 0) for k in keys)
    if le and ge: return Order.EQUAL
    if le:        return Order.BEFORE
    if ge:        return Order.AFTER
    return Order.CONCURRENT

def increment(clock: VC, node: NodeId) -> VC:
    return dict(clock) | {node: clock.get(node, 0) + 1}

def merge(a: VC, b: VC) -> VC:
    return {k: max(a.get(k, 0), b.get(k, 0)) for k in set(a) | set(b)}

লক্ষ্য করুন: le and ge একসাথে সত্য হলে সেটা EQUAL — দুই dict-এর key set ভিন্ন হলেও। Missing key মানেই ০, এই convention-টা পুরো library-তে এক জায়গাতেও ভাঙবেন না।

Checkpoint: ছয়টা জোড়া ঘড়ি হাতে বানিয়ে compare চালান — {a:1}, {a:1,b:1}, {b:1}, {}, {a:2}, {a:1,c:5} — প্রতিটার outcome আগে কাগজে লিখে, তারপর টেস্ট চালান। ভুল ধরা পড়বে এখনই, পরে নয়।

২. Compact JSON serialization

import json

def serialize(clock: VC) -> bytes:
    packed = {k: v for k, v in sorted(clock.items()) if v}
    return json.dumps(packed, separators=(",", ":")).encode()

def deserialize(data: bytes) -> VC:
    return {k: int(v) for k, v in json.loads(data).items()}

দুটো নিয়ম: zero entry কখনো wire-এ যাবে না (missing key মানেই ০, তাহলে ওগুলো বহন করাই অর্থহীন), আর key sort করতেই হবে — নইলে একই ঘড়ির দুই serialization ভিন্ন bytes হয়ে hash/test diff-এ বোমা ফাটায়।

Checkpoint: property test — যেকোনো ঘড়ির জন্য deserialize(serialize(c)) == c, আর serialize(c) deterministic (দুইবার চালিয়ে byte-equal)।

৩. Causal-buffer delivery gate

Causally ordered broadcast-এর নিয়ম: message m deliver করবেন শুধু তখন, যখন তার sender-এর আগের সব message পৌঁছে গেছে এবং m যে context-এ জন্মেছে সেই পূর্বসূরীগুলোও। ভেক্টর দিয়ে লেখা যায়:

deliverable(m) ⇔ vc[m][src] == local[src] + 1
                 এবং vc[m][k] <= local[k], সব k ≠ src-এর জন্য

class DeliveryGate:
    def __init__(self, node: NodeId):
        self.node = node
        self.clock: VC = {}
        self.held: list[dict] = []

    def _ready(self, m: dict) -> bool:
        vc, src = m["vc"], m["src"]
        if vc.get(src, 0) != self.clock.get(src, 0) + 1:
            return False
        return all(v <= self.clock.get(k, 0)
                   for k, v in vc.items() if k != src)

    def receive(self, m: dict) -> list[dict]:
        self.held.append(m)
        delivered = []
        progress = True
        while progress:                      # gap পূর্ণ হলে পিছনেরগুলোও ছাড়ে
            progress = False
            for m in list(self.held):
                if self._ready(m):
                    self.held.remove(m)
                    self.clock = merge(self.clock, m["vc"])
                    delivered.append(m)
                    progress = True
        return delivered

while-loop-টা আসল খেলা — একটা gap পূর্ণ হলে তার পেছনে আটকে থাকা সারি সারি message ছেড়ে দিতে হয়।

Checkpoint: ৩ node, ৫০ message, arrival order এলোমেলো করে shuffle করে feed করুন — delivered sequence প্রতিবার হুবহু এক হওয়াই প্রমাণ যে gate-টা causal delivery enforce করছে।

৪. Conflict-detector demo — shopping cart

দুই replica, একই base state, একসাথে দুই রকম edit:

base_cart = {"items": [], "clock": {}}

def add_item(replica_id, cart, item):
    new = dict(cart)
    new["items"] = cart["items"] + [item]
    new["clock"] = increment(cart["clock"], replica_id)
    return new

r1 = add_item("store-a", base_cart, "coffee")
r2 = add_item("store-b", base_cart, "tea")

verdict = compare(r1["clock"], r2["clock"])
assert verdict is Order.CONCURRENT

এখানেই মজা: last-write-wins নিলে দুইটার একটা নীরবে হারিয়ে যাবে — কোনো error ছাড়াই। সঠিক resolution হলো sibling রেখে union করা:

if verdict is Order.CONCURRENT:
    merged_items = sorted(set(r1["items"]) | set(r2["items"]))
    merged_clock = merge(r1["clock"], r2["clock"])

Demo script-টা দুই রকম resolver (lww বনাম union) চালিয়ে পাশাপাশি output দেখুন — lww-এর ক্ষেত্রে হারানো item-টা নিজের চোখে দেখা জরুরি।

Checkpoint: demo output-এ এই তিনটা লাইন থাকবে: CONCURRENT verdict, union result ["coffee", "tea"], আর lww result-এ একটা missing item।

৫. Fuzz harness — causality laws

import random

def fuzz(trials=10_000, seed=42):
    rng = random.Random(seed)
    nodes = ["n1", "n2", "n3"]
    for _ in range(trials):
        clocks = [{}, {}, {}]
        hist = []
        for _ in range(rng.randint(2, 20)):
            op = rng.choice(["inc", "merge"])
            i = rng.randrange(3)
            if op == "inc":
                clocks[i] = increment(clocks[i], nodes[i])
            else:
                j = rng.randrange(3)
                clocks[i] = merge(clocks[i], clocks[j])
            hist.append(dict(clocks[i]))

        for a in hist:
            for b in hist:
                ab, ba = compare(a, b), compare(b, a)
                assert not (ab is Order.BEFORE and ba is Order.BEFORE)   # antisymmetry
                if ab is Order.EQUAL:
                    assert ba is Order.EQUAL                             # symmetry

        for i in range(len(hist)):                                       # transitivity
            for j in range(len(hist)):
                for k in range(len(hist)):
                    ij, jk, ik = compare(hist[i], hist[j]), \
                                 compare(hist[j], hist[k]), \
                                 compare(hist[i], hist[k])
                    if ij is Order.BEFORE and jk is Order.BEFORE:
                        assert ik is Order.BEFORE

লক্ষ্য করুন কোন law কী বলছে: reflexivity — ঘড়ি নিজের সাথে EQUAL; antisymmetry — দুই পাশেই BEFORE অসম্ভব; symmetry — EQUAL দুই দিকেই; transitivity — BEFORE chain-এর প্রত্যক্ষ ফল। CONCURRENT-এর ক্ষেত্রে chain ইচ্ছাই ভাঙে, ওটা partial order-এর স্বভাব, bug নয়।

Checkpoint: seed বদলে ১০০ বার চালান; একবারও assertion না ভাঙলে laws ধরে নেওয়া যায়, ভাঙলে অভিনন্দন — fuzzer-ই আপনার প্রথম reviewer।

৬. Performance bench — 1e6 merges/sec

import timeit

clock_a, clock_b = ({f"n{i}": i for i in range(8)} for _ in range(2))
per_sec = 1_000_000 / (timeit.timeit(
    lambda: merge(clock_a, clock_b), number=200_000) / 200_000)
print(f"{per_sec:,.0f} merges/sec")

Naive dict version সাধারণত ২–৫ লক্ষ/sec-এ আটকায়। Profile করলে দেখবেন খরচ গিয়ে থাকে set union আর .get()-এ। দুটো চিরচেনা কৌশল:

• Fixed node universe হলে ঘড়িকে list[int]-এ নামান, node id → index mapping একবারই করা লাগে
• Merge-এ নতুন dict বানানোর বদলে in-place max লিখে reuse করুন (callers immutable ধরে চললে copy-on-write boundary এক জায়গায় রাখুন)

Checkpoint: ৮-entry ঘড়িতে ≥ 1,000,000 merges/sec sustained, warm-up বাদ দিয়ে median of 5 runs।

৭. API docs — compare outcome table

README-তে এই table-টা থাকতেই হবে — এটাই library-র আসল contract:

compare(a, b)	মানে	caller-এর করণীয়
EQUAL	একই জ্ঞান, দুই পাশে	কিছুই না
BEFORE	a আগে, b-র context-এ আছে	a বাদ, b গ্রহণ
AFTER	b আগে, a-র context-এ আছে	b বাদ, a গ্রহণ
CONCURRENT	ভিন্ন ইতিহাস, conflict	sibling রাখুন বা resolver চালান

প্রতিটা public function-এর docstring-এ একটা doctest রাখুন — এই library-র মতো pure data structure-এ doctest-ই সবচেয়ে সৎ ডকুমেন্টেশন।

নিজেকে চ্যালেঞ্জ করুন

1. Dotted version vector — ঘড়ির ভেতরের গরম entry-গুলো আলাদা রেখে serialization আরও ছোট করুন
2. Interval tree clocks — নতুন node join/leave-এ ঘড়ির size বাড়ে-কমে এমন ডিজাইন, fixed node id ছাড়া
3. Clock GC — চিরকাল চলা system-এ departed node-এর entry prune করার নিয়ম আর তার safety argument লিখুন
4. Binary wire format — varint + sorted keys, JSON-এর চেয়ে কত ছোট মাপুন
5. Hypothesis দিয়ে property-based test — নিজের fuzz harness-এর বিপরীতে দাঁড় করান
6. এই library-র উপরে একটা OR-Set CRDT বানান — add-wins semantics নিজেই

যা এখানে শিখলেন, পরে যেখানে লাগবে

এখানে	পরে
Partial-order compare	Consistency models — linearizability বনাম causal-এর পার্থক্য পড়া সহজ হবে
Causal delivery gate	Message queue-এ ordering guarantee-র ভেতরটা
Sibling + merge	Dynamo-ধাঁচা replicated KV store — এই track-এর পরের প্রজেক্টে সরাসরি লাগবে
Fuzz আর laws assert	যেকোনো protocol implementation-এর প্রথম প্রতিরক্ষা