Replicated KV Store

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

Quorum replication লেসনে আপনি R + W > N-এর অংক দেখেছেন। কিন্তু ওই অংকটা কাগজে থাকলে বোঝা যায় না সবচেয়ে জরুরি সত্যটা — quorum consistency-র চুক্তিটা শর্তসাপেক্ষ: message হারালে, replica মরলে, clock এদিক-ওদিক হলে চুক্তি রক্ষার ভার পড়ে repair, hint আর versioning- এর মতো যন্ত্রের কাঁধে। Dynamo paper-এর পুরো দৈর্ঘ্য আসলে এই যন্ত্রগুলোর বর্ণনা।

এই store-টা in-process — network loopback-এ তিনটা replica, fault injection একটা function call। হালকা হওয়ায় experiment দ্রুত চলে, আর experiment-ই এখানে মূল পাঠ: একই workload W/R-এর grid-এ চালিয়ে নিজের চোখে দেখা staleness কোথায় ঢুকে, কোন যন্ত্র কোন ফুটো বন্ধ করে।

ধাপে ধাপে

১. N-replica in-process store

class Replica:
    def __init__(self, name: str):
        self.name = name
        self.data: dict[str, list[VersionedValue]] = {}   # key -> siblings
        self.hints: list[tuple[str, str, VersionedValue]] = []
        self.alive = True

    def local_put(self, key: str, vv: VersionedValue):
        siblings = self.data.setdefault(key, [])
        # clock compare করে superseded বাদ, নতুন/CONCURRENT যোগ
        self.data[key] = reconcile(siblings, vv)

    def local_get(self, key: str) -> list[VersionedValue]:
        return self.data.get(key, [])

Coordinator (client-facing router) preference list বানায় — key অনুযায়ী N node-এর নির্দিষ্ট ক্রম:

def preference_list(key: str, nodes: list[Replica], n: int) -> list[Replica]:
    return sorted(nodes, key=lambda r: hash((key, r.name)))[:n]

Checkpoint: W=N, R=1 দিয়ে write-then-read — সবসময় নিজের লেখাই ফেরে; replica একটার data dict-এ হাত দিয়ে ভিন্ন value বসালে read-এ sibling দুটোই ফেরে — versioning ছাড়া এটা অসম্ভব, তাই ধাপ ৫-এর ভিত এখানেই।

২. Tunable W/R per request

def put(coordinator, key, value, w=2, timeout=0.5):
    vv = VersionedValue(value, increment(clock_for(key), coordinator.name),
                        wall=time.time())
    acks = gather_first([send_write(r, key, vv) for r in pref_list],
                        need=w, timeout=timeout)
    if len(acks) < w:
        raise TimeoutError(f"w={w} not met ({len(acks)})")
    return vv.clock

def get(coordinator, key, r=2, timeout=0.5):
    replies = gather_first([send_read(r_, key) for r_ in pref_list],
                           need=r, timeout=timeout)
    return reconcile_all(replies)      # সব sibling, তাজা ক্রমে

অংকটা হাতে খেলুন — N=3-তে তিনটা চরম config:

W	R	চরিত্র
3	1	write ধীর, read ঝটপট, এক replica গেলে write অসম্ভব
2	2	ভারসাম্য; এক failure সহ্য, overlap নিশ্চিত
1	3	write দ্রুত, read-heavy; এক replica গেলে read অসম্ভব

Checkpoint: এক replica kill করে প্রতিটা config-এ put/get চালান — কোনটা availability হারায়, কোনটা হারায় না, টেবিলে নোট করুন।

৩. Read-repair on stale hit

get চলাকালীন coordinator শুধু R-টা reply নয়, যত reply আসে সব তুলনা করে — stale replica-তে নতুন version ঠেলে দেয়:

def read_repair(replies: dict[str, list[VersionedValue]], key: str):
    newest = reconcile_all(list(replies.values()))
    for name, versions in replies.items():
        if any(compare(v.clock, n.clock) is Order.BEFORE
               for v in versions for n in newest):
            send_write(replica(name), key, newest)     # fire-and-forget

খরচটা read-এর উপর ছড়িয়ে যায় — background anti-entropy daemon-এর চেয়ে সস্তা, তবে শর্ত একটাই: যে key কেউ পড়ে না, সে কখনো repair হয় না। ফলে hot key সুস্থ থাকে, ভুলে যাওয়া key পচতে থাকে — এই সীমাটা dashboard-এ (ধাপ ৭) পরিষ্কার দেখা যাবে।

Checkpoint: W=1-এ write, এক replica-তে ইচ্ছে করে version পুরনো করুন, R=2 read — read-repair log-এ push line, পরের read-এ সমান।

৪. Hinted handoff

Preference list-এর node মরলে write ফেলে দেওয়া নয় — সুস্থ প্রতিবেশী hint হয়ে ধরে রাখে:

def send_write_with_hint(target: Replica, key: str, vv: VersionedValue,
                         home: Replica):
    if target.alive:
        target.local_put(key, vv)
    elif home.alive and home is not target:
        home.hints.append((target.name, key, vv))   # ঋণপত্র জমা
    else:
        raise Unavailable()

def replay_hints(node: Replica):                     # target ফিরলে
    remaining = []
    for tname, key, vv in node.hints:
        t = replica(tname)
        (t.local_put(key, vv) if t.alive else remaining.append(
            (tname, key, vv)))
    node.hints = remaining

Hint-এ target-এর নাম থাকে — data ওই node-এর উদ্দেশ্যেই, শুধু সাময়িক অতিথি। TTL না দিলে চিরতরে বন্ধ node-এর ঋণ জমে থাকে; lab-এ TTL ৫ মিনিট রাখুন আর backlog size metric-এ রাখুন।

Checkpoint: n2 মেরে W=2-তে দশটা write; n2 ফিরিয়ে replay — n2-এর data বাকিদের সমান; client কখনোই error দেখেনি।

৫. Versioned values + conflict UI

@dataclass
class VersionedValue:
    value: str
    clock: VC
    wall: float          # lww option-এর জন্যই, ordering-এর জন্য নয়

Concurrent write হলে get ফেরত দেয় sibling-দল — client UI সিদ্ধান্ত নেয়:

key: cart
[1] "coffee x2"   clock={a:4,b:1}
[2] "tea x1"      clock={a:1,b:4}
resolve (merge|lww|pick N)> merge
→ merged "coffee x2 + tea x1"  clock={a:4,b:4}

lww বেছে নিলে সিদ্ধান্ত হয় wall timestamp-এ — আর ঠিক সেখানেই clock-skew-এর ফাঁদ: এক replica-এর ঘড়ি ২ সেকেন্ড এগিয়ে দিন, তার পুরনো লেখা কীভাবে “তাজা” হয়ে নতুন লেখা গিলে ফেলে, demo-তে দেখান। Vector-clock merge আর lww-এর পার্থক্যটা এই UI-তেই হাতের অভিজ্ঞতা।

Checkpoint: skew demo-র output-এ দুই লাইন — skewed replica-এর value wins, আর তার wall ভবিষ্যতের তারিখ দেখাচ্ছে।

৬. Chaos runner

class ChaosLink:
    """Replica-দের মাঝে ঢোকানো নোংরা wire."""
    def __init__(self, seed=7):
        self.rng = random.Random(seed)
        self.drop_prob = 0.0

    def rpc(self, dst: Replica, op, *args):
        if self.rng.random() < self.drop_prob:
            self.dropped += 1
            raise MessageLost()          # caller retry বা timeout করবে
        return getattr(dst, op)(*args)

Runner matrix চালায় — drop_prob × W × R:

for p in (0.0, 0.05, 0.15, 0.30):
    for w, r in ((1, 3), (2, 2), (3, 1)):
        run_scenario(seed=100 + int(p * 100), drop=p, w=w, r=r)

Retry policy সহজ রাখুন — একবার retry, তারপর client-visible timeout. নইলে chaos-এর ফল নয়, retry storm-এর ফল মাপবেন।

Checkpoint: drop 30%-এ W=1 write-এর কত শতাংশ কোথাও পৌঁছায়নি — সংখ্যাটা নোটে; ধাপ ৭-এর dashboard-এ ওটাই staleness-এর উৎস।

৭. Consistency dashboard — staleness% sweep

Metric সংজ্ঞা কড়া রাখুন: staleness% = এমন read-এর ভগ্নাংশ যেগুলো আগে acknowledge হওয়া কোনো write-এর মান দেখেনি। Workload: ১০k write + ১০k read interleaved, প্রতি op-এ expected state ledger-এ রাখা।

staleness% (drop=10%)        W\R      1       2       3
                              1     18.7    11.2     6.9
                              2      8.4     2.6     1.1
                              3      3.9     0.4     0.0

Table-টা পড়ার তিনটা চোখ: R+W>N যত বেশি overlap, staleness তত নামে; কিন্তু drop বাড়লে quorum-এর গণিতও ভাঙে (W=1 লেখা কোথাও নাও পৌঁছাতে পারে); আর latency column পাশে রাখলে trade-off-এর পূর্ণ ছবি। Dashboard script-টা CSV-ও লিখবে — sweep-এর ফল পরের experiment-এর baseline।

Checkpoint: একই sweep read-repair বনাম ছাড়া দুইবার — পার্থক্যের column-ই read-repair-এর মূল্য নির্দিষ্ট সংখ্যায়।

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

1. Consistent hashing ring — preference list-এ virtual node, ring resize-এ migration কত কমে মাপুন
2. Merkle-tree anti-entropy — read না হওয়া key-ও repair হয় এমন background sync
3. Sloppy quorum — preference list ছোট হলে পাশের node-এ overflow, availability-র বিনিময়ে কী খরচ তা মাপুন
4. Tombstone আর delete GC — delete-ও একটা write, নইলে ফিরে আসে
5. Per-key timeout budget — coordinator deadline propagation-এ tail latency কত কমে
6. নিজের raft implementation project-এর cluster-কে coordinator-এর পেছনে metadata store হিসেবে বসান

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

এখানে	পরে
Quorum tuning intuition	Cassandra/DynamoDB consistency class বাছাই
Read-repair + hint	Anti-entropy design — যেকোনো eventually-consistent store
Sibling conflict UX	Offline-first app sync design
Staleness dashboard	Production SLA-তে “eventual” শব্দটার সংখ্যা-ভিত্তিক সংজ্ঞা