Raft Implementation

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

Raft-এর paper পড়ে “বুঝেছি” আর implement করে “বুঝিনি” — এই দুই অবস্থার মাঝখানের দূরত্ব মাপতে হলে একটাই উপায়, নিজে বানানো। Paper-এর প্রতিটা ছোট নিয়ম — votedFor persist করা, শেষ log entry-র term মিলিয়ে commit, RPC-তে term check — একেকটা আলাদা ভুল করার সুযোগ, আর প্রতিটা ভুলের শাস্তি একই: নীরব data loss বা দুই leader।

আর সবচেয়ে বড় প্রাপ্তি implementation-টা নয়, test harness-টা। Partition দিয়ে ক্লাস্টার দুই ভাগ করা, leader-কে replication-এর মাঝে মেরে ফেলা, তারপর history checker দিয়ে প্রমাণ করা যা হারায়নি — এই কায়দাটা এই প্রজেক্টের পর আপনি যেকোনো distributed system-এ কাজ করার সময় চালাতে পারবেন। etcd বা Consul-এর ভেতরের ইঞ্জিনটা এখন আর জাদুবাক্স থাকবে না।

ধাপে ধাপে

১. Node FSM ও persisted state

from dataclasses import dataclass, field, asdict
import json, os

@dataclass
class Persistent:
    current_term: int = 0
    voted_for: str | None = None
    log: list[dict] = field(default_factory=list)   # {term, cmd}

    def save(self, path: str):
        tmp = path + ".tmp"
        with open(tmp, "w") as f:
            json.dump(asdict(self), f)
            f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path)                        # atomic swap

তিনটা field persist না করলে কী হয়, এক লাইনে: দুর্ঘটনায় restart-এর পরে একই term-এ দুই candidate-কে vote দিলে একই term-এ দুই leader — Raft-এর মূল safety চুক্তিই ভাঙে। তাই vote দেওয়ার reply পাঠানোর আগে fsync, ক্রম উল্টানো যাবে না।

FSM হিসেবে রাখুন — role বদলায় শুধু নিয়ম বদলে:

follower : leader-এর heartbeat শুনছে, না পেলে candidate
candidate: নিজে vote নেয়, RequestVote ছড়ায়, majority পেলে leader
leader   : heartbeat + AppendEntries পাঠায়, client-এর লেখা গ্রহণ করে

Checkpoint: প্রতিটা role transition-এ একটা log line — n2 became candidate term=4। পুরো টেস্ট suite-এর প্রমাণ পরে এই lines থেকেই reconstruct করা যাবে।

২. RPC over threaded TCP bus

Frame সহজ রাখুন — length-prefix + JSON:

import struct, socket, threading, json

def send_msg(sock, obj):
    body = json.dumps(obj).encode()
    sock.sendall(struct.pack(">I", len(body)) + body)

def recv_exact(sock, n):
    buf = b""
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk: raise ConnectionError("closed")
        buf += chunk
    return buf

def recv_msg(sock):
    (n,) = struct.unpack(">I", recv_exact(sock, 4))
    return json.loads(recv_exact(sock, n))

প্রতি peer-এ reader thread; handler-এ প্রথম কাজ term check — নিয়ম একটাই: নিজের current_term-এর চেয়ে বড় term দেখলে সঙ্গে সঙ্গে follower-এ নেমে persist করা; ছোট term-এর RPC এসেছে বুঝি reply-তে নিজের term পাঠিয়ে reject। এই এক নিয়মই পুরনো leader-কে ফিরে এসে সরকার চালানো থেকে আটকায়।

Checkpoint: ৫-node cluster চালু করে হাতে একটা stale-term RequestVote পাঠান — reply-তে বড় term ফিরবে আর sender step down করবে।

৩. Randomized election timeout

ELECTION_MIN, ELECTION_MAX = 0.15, 0.30   # সেকেন্ড

def election_timeout(rng) -> float:
    return rng.uniform(ELECTION_MIN, ELECTION_MAX)

Timeout সবাই সমান হলে দুই candidate একসাথে দাঁড়িয়ে ভোট ভাগ হয়ে আবার একসাথে দাঁড়ায় — split vote চিরকাল। Randomization ওদের ভাঙে: একজন আগে জেগে উঠলে বাকিরা তাকে vote দিতে বাধ্য হয় (নিজে এই term-এ vote দিয়ে থাকলে)। Heartbeat পেলে timer reset — আর নিজের ভোটাধিকার এই term-এ খরচ হয়ে গেলে আর vote দেয় না।

Checkpoint: leader-কে SIGKILL → নতুন leader 1 সেকেন্ডের ভেতরে; ১০০ বার চালিয়ে একবারও একই term-এ দুই leader-এর log line নেই।

৪. Log replication — nextIndex ও conflict rewind

Leader প্রতি follower-এর জন্য nextIndex রাখে। AppendEntries-এর consistency check ব্যর্থ হলে (prevLogIndex-এ entry নেই বা term মেলে না) follower reject করে, leader nextIndex কমিয়ে আবার চেষ্টা:

def on_append_reject(self, peer, resp):
    ni = self.next_index[peer]
    if resp.get("conflict_term"):
        # নিজের log-এ ওই term-এর শেষ entry খুঁজে সেখানেই লাফান
        ct = resp["conflict_term"]
        idx = max((i for i, e in enumerate(self.log, 1)
                   if e["term"] == ct), default=resp["conflict_index"])
        self.next_index[peer] = min(idx + 1, ni) or 1
    else:
        self.next_index[peer] = max(ni - 1, 1)

Naive এক-এক করে নামাও চলে, কিন্তু conflict_term shortcut ছাড়া partition থেকে ফেরা follower-কে টানতে হাজার হাজার round-trip লাগে।

Checkpoint: follower-এর log হাতে কেটে ভিন্ন করে দিন (stale suffix জুড়ে দিন), restart করুন — leader ঠিক boundary থেকে overwrite করে মিলিয়ে দেবে, log শেষ পর্যন্ত পাঁচটা replica-তে byte-identical।

৫. Commit-index advance ও apply loop

def advance_commit(self):
    for n in range(len(self.log), self.commit_index, -1):
        if self.log[n - 1]["term"] != self.current_term:
            break                     # শুধু নিজের term-এর entry দিয়ে commit
        matched = 1 + sum(mi >= n for mi in self.match_index.values())
        if matched * 2 > len(self.peers) + 1:
            self.commit_index = n
            break

ওই term != current_term চেকটাই paper-এর Figure 8 — পুরনো term-এর entry leader replicate করে বসে থাকলেও নিজে থেকে commit করতে পারে না, নইলে ওই entry পরে overwrite হয়ে যেতে পারে। Apply loop আলাদা thread:

while True:
    if self.apply_index < self.commit_index:
        self.apply_index += 1
        self.kv[self.log[self.apply_index - 1]["cmd"]["k"]] = \
            self.log[self.apply_index - 1]["cmd"]["v"]

Checkpoint: ৫ node, দুইটা মেরে ফেলেও write/read চলে; মৃত দুটো ফিরে এসে catch-up করে; পাঁচটার kv state তুলনা করে equal।

৬. Partition-tolerant test harness

Network-কে abstraction-এর পেছনে লুকান — তাহলে test-এ partition মানেই একটা matrix flip:

class Bus:
    def __init__(self):
        self.alive = {n: True for n in NODES}
        self.groups = [{n for n in NODES}]     # ভাঙা নেই

    def partition(self, *groups):
        self.groups = [set(g) for g in groups]

    def reachable(self, a, b):
        return self.alive[a] and self.alive[b] and \
               any(a in g and b in g for g in self.groups)

Test-এর কায়দা:

• bus.partition(["n1","n2","n3"], ["n4","n5"]) — majority-তে election হয়ে write চলে, minority-র candidate চিরকাল timeout খায়, client-এর minority-তে পাঠানো write timeout ছাড়া কিছু পায় না
• Heal করলে বড় term-এর majority leader জেতে, minority-র সব state মুছে তার পেছনে চলে

Checkpoint: partition test-এ assert — minority-তে কোনো leader কখনো declare হয়নি, আর heal-এর পরে সব log identical।

৭. Leader crash mid-replication

সবচেয়ে সৎ test — replication-এর ঠিক মাঝপথে leader-কে মারুন:

def test_leader_crash_mid_replication(bus, nodes):
    leader = wait_for_leader(nodes)
    bus.drop_next_from(leader)              # ack যাওয়ার আগেই আটকান
    fut = client.submit_write("x=1")
    kill(leader)                            # 2/5-এ পৌঁছেছে, commit হয়নি
    new_leader = wait_for_leader(nodes)
    assert client.read("x") in (None, "1")  # uncommitted মানে অদৃশ্য হওয়াই ঠিক

যা প্রমাণ করতে হবে: commit হয়ে যাওয়া entry কখনোই হারায় না; commit হয়নি এমন entry হারলে সেটা দোষ নয় — client retry-ই protocol- এর চুক্তি। আর মৃত leader ফিরে এলে তার conflicting suffix ওই nextIndex rewind-এ কেটে যায় — ওই দৃশ্যের screenshot নোটে রাখুন।

Checkpoint: ৫০ random crash timing-এ চালান; history-তে একটাও committed-but-lost নেই।

৮. Jepsen-lite checker

Client driver প্রতিটা op-এর ইতিহাস রাখে:

{"op": ("write", "x", "7"), "invoke": t1, "ok": True, "complete": t2}
{"op": ("cas",  "x", "7", "8"), "invoke": t3, "ok": False}

Checker আলাদে চলে — history file পড়ে single-register linearizability যাচাই করে: প্রতিটা successful op-কে invoke-complete জানালার ভেতরের একটা বিন্দু ধরে real-time order-এ legal execution গড়া যায় কি না।

legal(write v)  : সবসময়
legal(read v)   : সর্বশেষ write/cas-এর ফলাফলই v
legal(cas a b)  : সর্বশেষ মান a হলে, নতুন মান b

প্রথম দিনে full checker না লিখে graph search ছোট করে শুরু করুন — op সংখ্যা কম হলে brute-force permutation-ই যথেষ্ট।

Checkpoint: ইচ্ছে করে একটা bug ঢোকান (commit চেকে term-এর শর্ত বাদ দিন) — checker যেন ধরে ফেলে; তারপর fix করে সবুজ করুন। Checker যেটা ধরতে পারে না, সে test লেখার কোনো অর্থ নেই।

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

1. Snapshotting + InstallSnapshot — লম্বা log কেটে ছোট করুন, catch-up এখন entry নয় snapshot বহন করবে
2. Pre-vote phase — বিচ্ছিন্ন node বারবার term বাড়িযে ফিরে এসে cluster-কে disturbance দেয়, সেটা আটকান
3. Read-only query — ReadIndex বা lease ছাড়া leader-এর read কেন stale হতে পারে, প্রমাণসহ লিখুন
4. Batch + pipeline AppendEntries — throughput কত বাড়ে মাপুন
5. Membership change — joint consensus বা single-server change যেকোনো একটা
6. Deterministic simulation — Bus আর scheduler-কে seeded করে একই interleaving replay করা

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

এখানে	পরে
Term + single-leader discipline	Kubernetes-এ etcd আর leader-election lease পড়া
Persist-then-reply কায়দা	যেকোনো WAL-ভিত্তিক database-এর durability চুক্তি
Partition harness	Production incident-এ “network partition?” প্রশ্নের সৎ উত্তর
History checker	Postmortem-এ “কী হারালো” প্রমাণসহ বলা