コンセンサスアルゴリズム・アンダーグラウンド【第2回】 ― クラッシュ故障の系譜、PaxosとRaft

class Acceptor:
    def __init__(self):
        self.promised = -1
        self.accepted_number = None
        self.accepted_value = None

    def prepare(self, number):
        if number <= self.promised:
            return None
        self.promised = number
        return self.accepted_number, self.accepted_value

    def accept(self, number, value):
        if number < self.promised:
            return False
        self.promised = number
        self.accepted_number = number
        self.accepted_value = value
        return True

responses = [a.prepare(n) for a in acceptors]
responses = [r for r in responses if r is not None]

if len(responses) >= quorum:
    previously_accepted = [r for r in responses if r[0] is not None]
    if previously_accepted:
        value = max(previously_accepted)[1]

    accepted = sum(a.accept(n, value) for a in acceptors)
    decided = accepted >= quorum

leader
 ├─ ACCEPT slot=101 value=A
 ├─ ACCEPT slot=102 value=B
 └─ ACCEPT slot=103 value=C

from dataclasses import dataclass

@dataclass
class Entry:
    term: int
    command: bytes

class RaftNode:
    def __init__(self, node_id, peers):
        self.node_id = node_id
        self.peers = peers
        self.state = "follower"
        self.current_term = 0
        self.voted_for = None
        self.log = []
        self.commit_index = -1
        self.last_applied = -1

def start_election(self):
    self.state = "candidate"
    self.current_term += 1
    self.voted_for = self.node_id
    votes = 1

    for peer in self.peers:
        response = peer.request_vote(
            term=self.current_term,
            candidate_id=self.node_id,
            last_log_index=len(self.log) - 1,
            last_log_term=self.log[-1].term if self.log else 0,
        )
        if response.term > self.current_term:
            self.become_follower(response.term)
            return
        votes += int(response.granted)

    if votes >= self.quorum():
        self.become_leader()

def request_vote(self, term, candidate_id, last_log_index, last_log_term):
    if term < self.current_term:
        return VoteResponse(self.current_term, False)

    if term > self.current_term:
        self.become_follower(term)

    local_term = self.log[-1].term if self.log else 0
    local_index = len(self.log) - 1

    up_to_date = (
        last_log_term > local_term
        or (last_log_term == local_term and last_log_index >= local_index)
    )

    can_vote = self.voted_for in (None, candidate_id)
    granted = can_vote and up_to_date

    if granted:
        self.voted_for = candidate_id
        self.persist_term_and_vote()

    return VoteResponse(self.current_term, granted)

def append_entries(self, term, prev_index, prev_term, entries, leader_commit):
    if term < self.current_term:
        return False

    if prev_index >= 0:
        if prev_index >= len(self.log):
            return False
        if self.log[prev_index].term != prev_term:
            return False

    insert_at = prev_index + 1
    for i, entry in enumerate(entries):
        pos = insert_at + i
        if pos < len(self.log) and self.log[pos].term != entry.term:
            self.log = self.log[:pos]
        if pos == len(self.log):
            self.log.append(entry)

    self.persist_log()

    if leader_commit > self.commit_index:
        self.commit_index = min(leader_commit, len(self.log) - 1)

    return True

view 10: primary=A
view 11: primary=C

zxid = (epoch, counter)

epoch / view / term
stable leader
ordered log
majority replication
leader replacement

Tokyo client → US leader → Tokyo replica

@dataclass
class Command:
    id: str
    reads: set[str]
    writes: set[str]

def conflicts(a: Command, b: Command) -> bool:
    return bool(
        a.writes & b.writes
        or a.writes & b.reads
        or b.writes & a.reads
    )

def dependencies(command, known_commands):
    return {
        other.id for other in known_commands
        if conflicts(command, other)
    }

Phase-1 quorum ∩ Phase-2 quorum ≠ ∅

range 1: A B C   leader=A
range 2: B C D   leader=C
range 3: C D E   leader=E

incorrect:            correct:
  send success          write WAL
  write WAL             fsync
                        send vote or success

batch = []
deadline = now() + MAX_DELAY

while len(batch) < MAX_BATCH and now() < deadline:
    command = try_receive()
    if command:
        batch.append(command)

append_and_fsync(batch)
replicate(batch)

if peer.inflight_bytes > MAX_INFLIGHT:
    pause_replication(peer)
else:
    send_append_entries(peer)

if applied_index - snapshot_index > SNAPSHOT_THRESHOLD:
    snapshot = state_machine.serialize()
    persist_snapshot(snapshot, applied_index)
    truncate_log(before=applied_index)

1. state machine snapshot を生成
2. snapshot metadata を永続化
3. snapshot が確実に読めることを確認
4. 対応するログを削除

C_old → C_old,new → C_new

for {
    select {
    case <-ticker.C:
        node.Tick()

    case msg := <-network:
        node.Step(ctx, msg)

    case ready := <-node.Ready():
        wal.Save(ready.HardState, ready.Entries)
        transport.Send(ready.Messages)
        applySnapshot(ready.Snapshot)
        applyCommitted(ready.CommittedEntries)
        node.Advance()
    }
}

Raft core
 ├─ durable WAL
 ├─ snapshot manager
 ├─ transport / TLS and identity
 ├─ message deduplication
 ├─ batching / backpressure
 ├─ membership changes
 ├─ read index / leader lease
 ├─ clock assumptions
 ├─ disk failure handling / corruption detection
 ├─ metrics / tracing / fault injection
 └─ rolling upgrade / state-machine versioning