ডিস্ট্রিবিউটেড কনসেনসাস
term ম্যানেজমেন্ট, vote request আর heartbeat সহ সরলীকৃত Raft leader election implement করুন।
গল্পে বুঝি
গ্রামের শালিস কমিটির সাত সদস্য বসেছে মেলার তারিখ ঠিক করতে। নিয়ম সোজা — কোনো সিদ্ধান্ত তখনই পাস, যখন সংখ্যাগরিষ্ঠ, মানে সাতজনের মধ্যে অন্তত চারজন একমত হয়। সেদিন ইবনে সিনা আর ফাতিমা আল-ফিহরি কাজে আটকে আসতে পারেনি, তবু বাকি পাঁচজন হাজির ছিল বলে বৈঠক থেমে থাকেনি — পাঁচজনের মধ্যে চারজন যেহেতু একটা তারিখে রাজি, তাই সিদ্ধান্ত পাকা হয়ে গেল, দুজন অনুপস্থিত থাকা সত্ত্বেও। প্রস্তাব কে তুলবে সেটা নিয়ে হুড়োহুড়ি এড়াতে ওরা আগেই আল-খোয়ারিজমিকে সভাপতি বেছে নিয়েছিল; সেই একাই প্রস্তাব তোলে, বাকিরা কেবল হ্যাঁ-না ভোট দেয়।
মাঝপথে আল-খোয়ারিজমিকে জরুরি কাজে গ্রামের বাইরে যেতে হলো, বৈঠক নেতৃত্বহীন। বাকিরা তখন আর অপেক্ষা না করে নতুন করে ভোট দিয়ে ফাতিমা আল-ফিহরিকে সভাপতি বানাল, আর কাজ যেভাবে চলছিল সেভাবেই চলতে থাকল। গুরুত্বপূর্ণ ব্যাপার — প্রতিটা চূড়ান্ত সিদ্ধান্ত সবাই একই কার্যবিবরণী খাতায় (মিনিট-বুকে) হুবহু একই কথায় লিখে রাখে, যাতে যে-ই পরে খাতা খুলুক, সবার কাছে ঠিক একই ইতিহাস থাকে, কোনো গরমিল নয়।
এই গল্পটাই আসলে distributed consensus। কমিটির সদস্যরা হলো node, সংখ্যাগরিষ্ঠের রাজি হওয়াটাই quorum — কয়েকজন অনুপস্থিত বা যোগাযোগহীন থাকলেও সিস্টেম আটকায় না। এক সভাপতি বেছে নেওয়া হলো leader election, আর সভাপতি হারিয়ে গেলে নতুন সভাপতি বেছে নেওয়াটাই re-election; সবার একই মিনিট-বুক হলো replicated log। Raft ঠিক এভাবেই কাজ করে, আর এর উপরেই দাঁড়িয়ে আছে etcd, ZooKeeper-এর মতো সিস্টেম — যেখানে একগুচ্ছ সার্ভারকে ব্যর্থতা সত্ত্বেও একটাই সিদ্ধান্তে একমত থাকতে হয়।
কনসেনসাস সমস্যা
একটা ডিস্ট্রিবিউটেড সিস্টেমে একাধিক সার্ভারকে একই মানে একমত হতে হয়, এমনকি যখন কিছু সার্ভার ক্র্যাশ করে বা নেটওয়ার্ক partition ঘটে। এটাই কনসেনসাস সমস্যা — আর এটা কম্পিউটার সায়েন্সের কঠিনতম সমস্যাগুলোর একটা।
Raft এটা সমাধান করে একটা মাত্র leader নির্বাচন করে, যে সব সিদ্ধান্ত নেয়। leader মারা গেলে বাকি node-গুলো একটা নতুন leader নির্বাচন করে। এটা Paxos-এর মতো বিকল্পগুলোর চেয়ে সহজ কিন্তু সমানভাবে সঠিক।
বাস্তব জীবনের উপমা
একটা board of directors-এর ভোটের মতো — একাধিক সদস্যকে একটা সিদ্ধান্তে একমত হতে হয়, আর পাস হতে সংখ্যাগরিষ্ঠতা লাগে। চেয়ারম্যান পদত্যাগ করলে একটা নতুন নির্বাচন হয়।
Raft-এর মূল ধারণা
- Term — একটা লজিক্যাল ঘড়ি। প্রতিটা নির্বাচন term বাড়িয়ে দেয়। উচ্চতর term থাকা node বিরোধে জেতে।
- Leader — সব ক্লায়েন্ট রিকোয়েস্ট হ্যান্ডল করে আর follower-দের কাছে replicate করে
- Candidate — একটা follower যে leader-এর কাছ থেকে কিছু শোনেনি আর একটা নির্বাচন শুরু করে
- Follower — passive node যে leader/candidate-এর রিকোয়েস্টে সাড়া দেয়
- Majority (quorum) — leader হতে একটা candidate-কে সংখ্যাগরিষ্ঠ ভোট লাগে (যেমন 5-এর মধ্যে 3)
import dgram from 'node:dgram';
// --- Types ---
type NodeState = 'follower' | 'candidate' | 'leader';
interface VoteRequest {
type: 'vote_request';
term: number;
candidateId: string;
lastLogIndex: number;
lastLogTerm: number;
}
interface VoteResponse {
type: 'vote_response';
term: number;
voteGranted: boolean;
voterId: string;
}
interface Heartbeat {
type: 'heartbeat';
term: number;
leaderId: string;
}
interface HeartbeatResponse {
type: 'heartbeat_response';
term: number;
nodeId: string;
success: boolean;
}
type RaftMessage = VoteRequest | VoteResponse | Heartbeat | HeartbeatResponse;
interface Peer {
id: string;
host: string;
port: number;
}
// --- Raft Node ---
class RaftNode {
private state: NodeState = 'follower';
private currentTerm = 0;
private votedFor: string | null = null;
private leaderId: string | null = null;
private votesReceived = new Set<string>();
private electionTimeout: ReturnType<typeof setTimeout> | null = null;
private heartbeatInterval: ReturnType<typeof setInterval> | null = null;
private socket: dgram.Socket;
// Timing (randomized to prevent split votes)
private readonly ELECTION_TIMEOUT_MIN = 1500; // ms
private readonly ELECTION_TIMEOUT_MAX = 3000;
private readonly HEARTBEAT_INTERVAL = 500;
constructor(
private readonly id: string,
private readonly port: number,
private readonly peers: Peer[],
private readonly logIndex: number = 0,
private readonly logTerm: number = 0
) {
this.socket = dgram.createSocket('udp4');
this.socket.on('message', (data) => this.handleMessage(data));
this.socket.bind(port);
}
start(): void {
console.log(`[${this.id}] Starting as follower (term ${this.currentTerm})`);
this.resetElectionTimeout();
}
private randomElectionTimeout(): number {
return (
this.ELECTION_TIMEOUT_MIN +
Math.random() * (this.ELECTION_TIMEOUT_MAX - this.ELECTION_TIMEOUT_MIN)
);
}
private resetElectionTimeout(): void {
if (this.electionTimeout) clearTimeout(this.electionTimeout);
this.electionTimeout = setTimeout(() => this.startElection(), this.randomElectionTimeout());
}
// --- State transitions ---
private startElection(): void {
this.state = 'candidate';
this.currentTerm++;
this.votedFor = this.id;
this.votesReceived.clear();
this.votesReceived.add(this.id); // vote for self
console.log(`[${this.id}] Starting election for term ${this.currentTerm}`);
// Request votes from all peers
const request: VoteRequest = {
type: 'vote_request',
term: this.currentTerm,
candidateId: this.id,
lastLogIndex: this.logIndex,
lastLogTerm: this.logTerm
};
for (const peer of this.peers) {
this.send(peer, request);
}
this.resetElectionTimeout();
}
private becomeLeader(): void {
if (this.state !== 'candidate') return;
this.state = 'leader';
this.leaderId = this.id;
console.log(`[${this.id}] Became LEADER for term ${this.currentTerm}`);
// Stop election timer
if (this.electionTimeout) clearTimeout(this.electionTimeout);
// Start sending heartbeats
this.sendHeartbeats();
this.heartbeatInterval = setInterval(() => this.sendHeartbeats(), this.HEARTBEAT_INTERVAL);
}
private stepDown(newTerm: number): void {
console.log(`[${this.id}] Stepping down. Old term: ${this.currentTerm}, new term: ${newTerm}`);
this.state = 'follower';
this.currentTerm = newTerm;
this.votedFor = null;
this.leaderId = null;
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.heartbeatInterval = null;
}
this.resetElectionTimeout();
}
// --- Message handling ---
private handleMessage(data: Buffer): void {
const msg: RaftMessage = JSON.parse(data.toString());
switch (msg.type) {
case 'vote_request':
this.handleVoteRequest(msg);
break;
case 'vote_response':
this.handleVoteResponse(msg);
break;
case 'heartbeat':
this.handleHeartbeat(msg);
break;
case 'heartbeat_response':
this.handleHeartbeatResponse(msg);
break;
}
}
private handleVoteRequest(req: VoteRequest): void {
// If request has higher term, step down
if (req.term > this.currentTerm) {
this.stepDown(req.term);
}
let granted = false;
if (
req.term >= this.currentTerm &&
(this.votedFor === null || this.votedFor === req.candidateId) &&
(req.lastLogTerm > this.logTerm ||
(req.lastLogTerm === this.logTerm && req.lastLogIndex >= this.logIndex))
) {
granted = true;
this.votedFor = req.candidateId;
this.resetElectionTimeout(); // reset since we granted a vote
}
const response: VoteResponse = {
type: 'vote_response',
term: this.currentTerm,
voteGranted: granted,
voterId: this.id
};
const peer = this.peers.find((p) => p.id === req.candidateId);
if (peer) this.send(peer, response);
}
private handleVoteResponse(res: VoteResponse): void {
if (res.term > this.currentTerm) {
this.stepDown(res.term);
return;
}
if (this.state !== 'candidate' || res.term !== this.currentTerm) return;
if (res.voteGranted) {
this.votesReceived.add(res.voterId);
const majority = Math.floor((this.peers.length + 1) / 2) + 1;
console.log(
`[${this.id}] Got vote from ${res.voterId} (${this.votesReceived.size}/${majority} needed)`
);
if (this.votesReceived.size >= majority) {
this.becomeLeader();
}
}
}
private handleHeartbeat(hb: Heartbeat): void {
if (hb.term >= this.currentTerm) {
if (this.state !== 'follower') {
this.stepDown(hb.term);
}
this.currentTerm = hb.term;
this.leaderId = hb.leaderId;
this.resetElectionTimeout();
}
const peer = this.peers.find((p) => p.id === hb.leaderId);
if (peer) {
const response: HeartbeatResponse = {
type: 'heartbeat_response',
term: this.currentTerm,
nodeId: this.id,
success: hb.term >= this.currentTerm
};
this.send(peer, response);
}
}
private handleHeartbeatResponse(_res: HeartbeatResponse): void {
// In a full implementation, track which followers are up-to-date
}
// --- Communication ---
private sendHeartbeats(): void {
const hb: Heartbeat = {
type: 'heartbeat',
term: this.currentTerm,
leaderId: this.id
};
for (const peer of this.peers) {
this.send(peer, hb);
}
}
private send(peer: Peer, msg: RaftMessage): void {
const data = Buffer.from(JSON.stringify(msg));
this.socket.send(data, peer.port, peer.host);
}
getStatus(): object {
return {
id: this.id,
state: this.state,
term: this.currentTerm,
leader: this.leaderId,
votedFor: this.votedFor
};
}
}
// --- Run a node ---
const nodeId = process.argv[2] || 'node-1';
const port = parseInt(process.argv[3] || '4001');
const allNodes: Peer[] = [
{ id: 'node-1', host: '127.0.0.1', port: 4001 },
{ id: 'node-2', host: '127.0.0.1', port: 4002 },
{ id: 'node-3', host: '127.0.0.1', port: 4003 }
];
const peers = allNodes.filter((n) => n.id !== nodeId);
const node = new RaftNode(nodeId, port, peers);
node.start();
// Print status periodically
setInterval(() => {
console.log(`[${nodeId}] Status:`, node.getStatus());
}, 5000);
process.on('SIGINT', () => process.exit(0));package main
import (
"encoding/json"
"fmt"
"log"
"math/rand"
"net"
"os"
"sync"
"time"
)
// --- Types ---
type NodeState int
const (
Follower NodeState = iota
Candidate
Leader
)
func (s NodeState) String() string {
return [...]string{"follower", "candidate", "leader"}[s]
}
type VoteRequest struct {
Type string `json:"type"`
Term int `json:"term"`
CandidateID string `json:"candidateId"`
LastLogIndex int `json:"lastLogIndex"`
LastLogTerm int `json:"lastLogTerm"`
}
type VoteResponse struct {
Type string `json:"type"`
Term int `json:"term"`
VoteGranted bool `json:"voteGranted"`
VoterID string `json:"voterId"`
}
type Heartbeat struct {
Type string `json:"type"`
Term int `json:"term"`
LeaderID string `json:"leaderId"`
}
type Peer struct {
ID string
Addr *net.UDPAddr
}
// --- Raft Node ---
type RaftNode struct {
mu sync.Mutex
id string
state NodeState
currentTerm int
votedFor string
leaderID string
votesReceived map[string]bool
logIndex int
logTerm int
peers []*Peer
conn *net.UDPConn
electionTimer *time.Timer
heartbeatTicker *time.Ticker
stopCh chan struct{}
}
const (
electionTimeoutMin = 1500 * time.Millisecond
electionTimeoutMax = 3000 * time.Millisecond
heartbeatInterval = 500 * time.Millisecond
)
func NewRaftNode(id string, port int, peers []*Peer) (*RaftNode, error) {
addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf(":%d", port))
if err != nil {
return nil, err
}
conn, err := net.ListenUDP("udp", addr)
if err != nil {
return nil, err
}
return &RaftNode{
id: id,
state: Follower,
votesReceived: make(map[string]bool),
peers: peers,
conn: conn,
stopCh: make(chan struct{}),
}, nil
}
func randomTimeout() time.Duration {
return electionTimeoutMin +
time.Duration(rand.Int63n(int64(electionTimeoutMax-electionTimeoutMin)))
}
func (n *RaftNode) Start() {
log.Printf("[%s] Starting as follower (term %d)", n.id, n.currentTerm)
n.resetElectionTimer()
go n.receiveMessages()
}
func (n *RaftNode) resetElectionTimer() {
if n.electionTimer != nil {
n.electionTimer.Stop()
}
n.electionTimer = time.AfterFunc(randomTimeout(), func() {
n.mu.Lock()
defer n.mu.Unlock()
n.startElection()
})
}
func (n *RaftNode) startElection() {
n.state = Candidate
n.currentTerm++
n.votedFor = n.id
n.votesReceived = map[string]bool{n.id: true}
log.Printf("[%s] Starting election for term %d", n.id, n.currentTerm)
req := VoteRequest{
Type: "vote_request", Term: n.currentTerm,
CandidateID: n.id, LastLogIndex: n.logIndex, LastLogTerm: n.logTerm,
}
for _, peer := range n.peers {
n.sendTo(peer, req)
}
n.resetElectionTimer()
}
func (n *RaftNode) becomeLeader() {
if n.state != Candidate {
return
}
n.state = Leader
n.leaderID = n.id
log.Printf("[%s] Became LEADER for term %d", n.id, n.currentTerm)
if n.electionTimer != nil {
n.electionTimer.Stop()
}
// Start heartbeats
n.sendHeartbeats()
n.heartbeatTicker = time.NewTicker(heartbeatInterval)
go func() {
for range n.heartbeatTicker.C {
n.mu.Lock()
if n.state == Leader {
n.sendHeartbeats()
}
n.mu.Unlock()
}
}()
}
func (n *RaftNode) stepDown(newTerm int) {
log.Printf("[%s] Stepping down. Old term: %d, new: %d", n.id, n.currentTerm, newTerm)
n.state = Follower
n.currentTerm = newTerm
n.votedFor = ""
n.leaderID = ""
if n.heartbeatTicker != nil {
n.heartbeatTicker.Stop()
n.heartbeatTicker = nil
}
n.resetElectionTimer()
}
// --- Message handling ---
func (n *RaftNode) receiveMessages() {
buf := make([]byte, 4096)
for {
size, _, err := n.conn.ReadFromUDP(buf)
if err != nil {
continue
}
var raw map[string]interface{}
json.Unmarshal(buf[:size], &raw)
msgType, _ := raw["type"].(string)
n.mu.Lock()
switch msgType {
case "vote_request":
var req VoteRequest
json.Unmarshal(buf[:size], &req)
n.handleVoteRequest(req)
case "vote_response":
var res VoteResponse
json.Unmarshal(buf[:size], &res)
n.handleVoteResponse(res)
case "heartbeat":
var hb Heartbeat
json.Unmarshal(buf[:size], &hb)
n.handleHeartbeat(hb)
}
n.mu.Unlock()
}
}
func (n *RaftNode) handleVoteRequest(req VoteRequest) {
if req.Term > n.currentTerm {
n.stepDown(req.Term)
}
granted := false
if req.Term >= n.currentTerm &&
(n.votedFor == "" || n.votedFor == req.CandidateID) &&
(req.LastLogTerm > n.logTerm ||
(req.LastLogTerm == n.logTerm && req.LastLogIndex >= n.logIndex)) {
granted = true
n.votedFor = req.CandidateID
n.resetElectionTimer()
}
resp := VoteResponse{
Type: "vote_response", Term: n.currentTerm,
VoteGranted: granted, VoterID: n.id,
}
for _, p := range n.peers {
if p.ID == req.CandidateID {
n.sendTo(p, resp)
break
}
}
}
func (n *RaftNode) handleVoteResponse(res VoteResponse) {
if res.Term > n.currentTerm {
n.stepDown(res.Term)
return
}
if n.state != Candidate || res.Term != n.currentTerm {
return
}
if res.VoteGranted {
n.votesReceived[res.VoterID] = true
majority := (len(n.peers)+1)/2 + 1
log.Printf("[%s] Got vote from %s (%d/%d)",
n.id, res.VoterID, len(n.votesReceived), majority)
if len(n.votesReceived) >= majority {
n.becomeLeader()
}
}
}
func (n *RaftNode) handleHeartbeat(hb Heartbeat) {
if hb.Term >= n.currentTerm {
if n.state != Follower {
n.stepDown(hb.Term)
}
n.currentTerm = hb.Term
n.leaderID = hb.LeaderID
n.resetElectionTimer()
}
}
func (n *RaftNode) sendHeartbeats() {
hb := Heartbeat{Type: "heartbeat", Term: n.currentTerm, LeaderID: n.id}
for _, peer := range n.peers {
n.sendTo(peer, hb)
}
}
func (n *RaftNode) sendTo(peer *Peer, msg interface{}) {
data, _ := json.Marshal(msg)
n.conn.WriteToUDP(data, peer.Addr)
}
func main() {
nodeID := "node-1"
port := 4001
if len(os.Args) > 1 {
nodeID = os.Args[1]
}
if len(os.Args) > 2 {
fmt.Sscanf(os.Args[2], "%d", &port)
}
allNodes := map[string]int{"node-1": 4001, "node-2": 4002, "node-3": 4003}
var peers []*Peer
for id, p := range allNodes {
if id != nodeID {
addr, _ := net.ResolveUDPAddr("udp", fmt.Sprintf("127.0.0.1:%d", p))
peers = append(peers, &Peer{ID: id, Addr: addr})
}
}
node, err := NewRaftNode(nodeID, port, peers)
if err != nil {
log.Fatal(err)
}
node.Start()
// Print status
ticker := time.NewTicker(5 * time.Second)
for range ticker.C {
node.mu.Lock()
log.Printf("[%s] state=%s term=%d leader=%s",
node.id, node.state, node.currentTerm, node.leaderID)
node.mu.Unlock()
}
}মূল শিক্ষা
- Raft নিশ্চিত করে প্রতি term-এ শুধু একটাই leader থাকে — সংখ্যাগরিষ্ঠ ভোট বাধ্যতামূলক করে split-brain ঠেকানো হয়
- Randomized election timeout একাধিক node-এর একসাথে নির্বাচন শুরু করা ঠেকায়
- উচ্চতর term সবসময় জেতে — একটা node উচ্চতর term দেখলে সাথে সাথে step down করে
- একটা candidate-কে vote পেতে হলে up-to-date log থাকতে হবে — এটা বাসি node-কে leader হওয়া থেকে ঠেকায়
- leader-এর কাছ থেকে heartbeat অপ্রয়োজনীয় নির্বাচন ঠেকায়
বাস্তব ব্যবহার
- etcd (Kubernetes ব্যবহার করে) ডিস্ট্রিবিউটেড key-value কনসেনসাসের জন্য Raft implement করে
- CockroachDB node জুড়ে consistent replication-এর জন্য Raft ব্যবহার করে
- HashiCorp Consul service discovery কনসেনসাসের জন্য Raft ব্যবহার করে
- আপনি নিজে খুব কমই কনসেনসাস implement করবেন। etcd, ZooKeeper, বা Consul ব্যবহার করুন। কিন্তু Raft বোঝা এই সিস্টেমগুলো ডিবাগ আর অপারেট করতে সাহায্য করে।