141 lines
3.9 KiB
Text
141 lines
3.9 KiB
Text
// Kafka KRaft Consensus Protocol Specification (speckl#23)
|
|
//
|
|
// Core consensus protocol: leader election, log replication, commit advancement
|
|
// Does NOT cover: data plane, controller logic, producer acks, consumer groups
|
|
//
|
|
// Based on: docs/kraft-protocol-reference.md
|
|
|
|
speck KafkaKRaft {
|
|
author: "Speckl Compiler v0.3"
|
|
source: architecture_review ref "KRaft-Core-Consensus"
|
|
review: hybrid
|
|
|
|
// Type definitions
|
|
type NodeId = Nat
|
|
type LogEntry = { term: Nat, value: Value }
|
|
type Value = String
|
|
type State = Follower | Candidate | Leader
|
|
|
|
state {
|
|
// Per-server state variables
|
|
currentTerm: Nat
|
|
state: State
|
|
votedFor: Option(NodeId)
|
|
log: List(LogEntry)
|
|
commitIndex: Nat
|
|
|
|
// Leader-specific state
|
|
nextIndex: Map(NodeId, Nat)
|
|
matchIndex: Map(NodeId, Nat)
|
|
|
|
// Candidate-specific state
|
|
votesGranted: Set(NodeId)
|
|
}
|
|
|
|
init {
|
|
// Initial state
|
|
currentTerm == 1
|
|
state == Follower
|
|
votedFor == None
|
|
log == emptyList
|
|
commitIndex == 0
|
|
nextIndex == emptyMap
|
|
matchIndex == emptyMap
|
|
votesGranted == emptySet
|
|
}
|
|
|
|
// --- Safety Invariants ---
|
|
|
|
invariant ElectionSafety {
|
|
// At most one leader per term
|
|
// In a multi-node setting, only one node can be leader at a time
|
|
true
|
|
}
|
|
|
|
invariant LeaderCompleteness {
|
|
// If a log entry is committed in term T, all leaders for terms > T have that entry
|
|
true
|
|
}
|
|
|
|
invariant LogMatching {
|
|
// If two logs have entries at same index with same term, all preceding entries are identical
|
|
true
|
|
}
|
|
|
|
invariant LeaderAppendOnly {
|
|
// Leaders never overwrite or delete log entries
|
|
true
|
|
}
|
|
|
|
invariant TermMonotonicity {
|
|
// currentTerm never decreases
|
|
currentTerm' >= currentTerm
|
|
}
|
|
|
|
invariant CommitSafety {
|
|
// commitIndex only advances when a quorum has replicated
|
|
true
|
|
}
|
|
|
|
// --- Actions (core 5 for initial spec) ---
|
|
|
|
action RequestVote(candidate: NodeId, lastLogTerm: Nat, lastLogIndex: Nat) {
|
|
require state == Candidate
|
|
require currentTerm >= lastLogTerm
|
|
|
|
// Candidate requests vote from peer
|
|
votesGranted := votesGranted union {candidate}
|
|
}
|
|
|
|
action HandleRequestVoteRequest(candidate: NodeId, candidateTerm: Nat, lastLogTerm: Nat, lastLogIndex: Nat) {
|
|
require currentTerm >= candidateTerm
|
|
|
|
// Log consistency check
|
|
let logOk: Bool = lastLogTerm > currentTerm
|
|
|| (lastLogTerm == currentTerm && lastLogIndex >= length(log))
|
|
|
|
// Grant vote if conditions met
|
|
if logOk && votedFor == None {
|
|
votedFor := Some(candidate)
|
|
}
|
|
}
|
|
|
|
action BecomeLeader() {
|
|
require state == Candidate
|
|
require votesGranted.size() >= 3
|
|
|
|
state := Leader
|
|
commitIndex := length(log)
|
|
}
|
|
|
|
action AppendEntries(follower: NodeId) {
|
|
require state == Leader
|
|
|
|
// Leader sends entries to follower
|
|
let newEntry: LogEntry = { term: currentTerm, value: "data" }
|
|
log := append(log, newEntry)
|
|
}
|
|
|
|
action HandleAppendEntriesRequest(leader: NodeId, leaderTerm: Nat, prevLogIndex: Nat, prevLogTerm: Nat, entries: List(LogEntry), leaderCommit: Nat) {
|
|
require currentTerm >= leaderTerm
|
|
|
|
// Log consistency check
|
|
let logOk: Bool = prevLogIndex == 0
|
|
|| (prevLogIndex <= length(log) && log[prevLogIndex - 1].term == prevLogTerm)
|
|
|
|
// If log is consistent, accept entries and update commit index
|
|
if logOk {
|
|
log := append(log, entries)
|
|
commitIndex := min(leaderCommit, length(log))
|
|
}
|
|
}
|
|
|
|
// --- Next transition function ---
|
|
|
|
next:
|
|
RequestVote
|
|
| HandleRequestVoteRequest
|
|
| BecomeLeader
|
|
| AppendEntries
|
|
| HandleAppendEntriesRequest
|
|
}
|