diff --git a/docs/kraft-protocol-reference.md b/docs/kraft-protocol-reference.md new file mode 100644 index 0000000..d301a3b --- /dev/null +++ b/docs/kraft-protocol-reference.md @@ -0,0 +1,318 @@ +# Kafka KRaft Protocol Reference + +> Source material for writing `examples/KafkaKRaft.speck` (speckl#22) + +## Scope + +This reference covers the **consensus core** of KRaft — the state machine for leader election, log replication, and commit advancement. It does NOT cover: + +- Full broker functionality (topic creation, partition assignment, consumer group coordination) +- Data plane (produce/fetch request handling) +- Controller-specific logic (metadata quarums, configuration management) +- Network transport (this is a logical model, not a network protocol spec) + +The boundary: if it touches `currentTerm`, `votedFor`, `log`, `commitIndex`, or `state` transitions, it's in scope. + +## Sources + +1. **Raft TLA+ Specification** — Diego Ongaro, 2014. The canonical formal spec. Available at: `github.com/ongardie/raft.tla` +2. **"In Search of an Understandable Consensus Algorithm"** — Ongaro & Ousterhout, 2014 (USENIX ATC). Includes safety proof. +3. **KIP-500: Next Generation Client Protocol** — KRaft design document. URL changes; search Apache Kafka JIRA for KIP-500. +4. **KIP-642: Dynamic Configuration in KRaft** — Configuration updates in KRaft mode. +5. **KIP-830: KRaft Metadata Quorum** — The actual metadata quorum implementation. +6. **Apache Kafka source** — `kafka/raft/` package in the Apache Kafka repository. + +## Raft State Machine (from TLA+ Spec) + +### State Variables + +Per server (`i ∈ Server`): + +| Variable | Type | Description | +|----------|------|-------------| +| `currentTerm` | Nat | Current term number (monotonically increasing) | +| `state` | {Follower, Candidate, Leader} | Server role | +| `votedFor` | Server ∪ {Nil} | Candidate voted for in current term, or Nil | +| `log` | Seq(LogEntry) | Sequence of log entries (index = position) | +| `commitIndex` | Nat | Index of highest log entry known to be committed | +| `nextIndex` | Server → Nat | (Leader only) Next log index to send to each follower | +| `matchIndex` | Server → Nat | (Leader only) Highest log index replicated on each follower | +| `votesResponded` | Set(Server) | (Candidate only) Servers that responded to RequestVote | +| `votesGranted` | Set(Server) | (Candidate only) Servers that granted vote | + +Global: + +| Variable | Type | Description | +|----------|------|-------------| +| `messages` | Bag(Message) | In-flight messages (requests/responses) | + +### LogEntry Type + +``` +LogEntry == { term: Nat, value: Value } +``` + +### Quorum + +``` +Quorum == {i ∈ SUBSET(Server) : Cardinality(i) * 2 > Cardinality(Server)} +``` + +Simple majority. The key property: any two quorums overlap. + +### Initial State + +``` +currentTerm[i] = 1 +state[i] = Follower +votedFor[i] = Nil +log[i] = <<>> (empty sequence) +commitIndex[i] = 0 +``` + +### Actions (State Transitions) + +#### 1. Timeout → Start Election + +**Preconditions:** +- `state[i] ∈ {Follower, Candidate}` + +**Effects:** +- `state[i] ← Candidate` +- `currentTerm[i] ← currentTerm[i] + 1` +- `votedFor[i] ← Nil` +- `votesResponded[i] ← {}` +- `votesGranted[i] ← {}` + +#### 2. RequestVote Request (Candidate → Peer) + +**Preconditions:** +- `state[i] = Candidate` +- `j ∉ votesResponded[i]` + +**Effects:** +- Send `RequestVoteRequest` with `{mterm: currentTerm[i], mlastLogTerm: LastTerm(log[i]), mlastLogIndex: Len(log[i]), msource: i, mdest: j}` + +#### 3. HandleRequestVoteRequest (Peer → Candidate) + +**Preconditions:** +- `m.mterm ≤ currentTerm[i]` + +**Log check (`logOk`):** +``` +logOk ≡ m.mlastLogTerm > LastTerm(log[i]) + ∨ (m.mlastLogTerm = LastTerm(log[i]) ∧ m.mlastLogIndex ≥ Len(log[i])) +``` + +**Grant conditions:** +``` +grant ≡ m.mterm = currentTerm[i] ∧ logOk ∧ votedFor[i] ∈ {Nil, j} +``` + +**Effects (if grant):** +- `votedFor[i] ← j` + +**Effects (always):** +- Reply with `RequestVoteResponse{mterm: currentTerm[i], mvoteGranted: grant}` + +#### 4. HandleRequestVoteResponse (Candidate receives vote) + +**Preconditions:** +- `m.mterm = currentTerm[i]` + +**Effects:** +- `votesResponded[i] ← votesResponded[i] ∪ {j}` +- If `m.mvoteGranted`: `votesGranted[i] ← votesGranted[i] ∪ {j}` + +#### 5. BecomeLeader (Candidate → Leader) + +**Preconditions:** +- `state[i] = Candidate` +- `votesGranted[i] ∈ Quorum` + +**Effects:** +- `state[i] ← Leader` +- `nextIndex[i] ← [j ∈ Server → Len(log[i]) + 1]` for all j +- `matchIndex[i] ← [j ∈ Server → 0]` for all j + +#### 6. AppendEntries Request (Leader → Follower) + +**Preconditions:** +- `i ≠ j` +- `state[i] = Leader` + +**Effects:** +- Send `AppendEntriesRequest` with entries from `nextIndex[i][j]` to `min(Len(log[i]), nextIndex[i][j])` (at most 1 entry per message in the spec) +- Include `mprevLogIndex = nextIndex[i][j] - 1`, `mprevLogTerm`, `mcommitIndex` + +#### 7. HandleAppendEntriesRequest (Follower receives entries) + +**Preconditions:** +- `m.mterm ≤ currentTerm[i]` + +**Log consistency check (`logOk`):** +``` +logOk ≡ m.mprevLogIndex = 0 + ∨ (m.mprevLogIndex ≤ Len(log[i]) ∧ m.mprevLogTerm = log[i][m.mprevLogIndex].term) +``` + +**Effects (if logOk AND m.mterm ≥ currentTerm[i]):** +- Reset to Follower if Candidate +- Update `currentTerm[i] ← m.mterm` if needed +- Append/conflict resolution: if `m.mprevLogIndex > 0`, entries after `mprevLogIndex` that conflict are truncated +- `commitIndex[i] ← min(m.mcommitIndex, Len(log[i]))` +- Reply `AppendEntriesResponse{msuccess: TRUE, mmatchIndex: ...}` + +**Effects (if !logOk OR m.mterm < currentTerm[i]):** +- Reply `AppendEntriesResponse{msuccess: FALSE, mmatchIndex: 0}` + +#### 8. AdvanceCommitIndex (Leader) + +**Preconditions:** +- `state[i] = Leader` + +**Effects:** +- Find highest `N` such that `N > commitIndex[i]` and a quorum of `matchIndex[i][k] ≥ N` +- Only commit entries from current term: `log[i][N].term = currentTerm[i]` +- `commitIndex[i] ← N` + +#### 9. ClientRequest (Leader appends entry) + +**Preconditions:** +- `state[i] = Leader` + +**Effects:** +- Append `[term: currentTerm[i], value: v]` to `log[i]` + +#### 10. Restart (Server crash recovery) + +**Effects:** +- `state[i] ← Follower` +- Reset volatile state (`votesResponded`, `votesGranted`, `nextIndex`, `matchIndex`, `commitIndex`) +- Preserve persistent state (`currentTerm`, `votedFor`, `log`) + +### Safety Invariants (from Raft paper) + +These are the invariants we need to prove in SpeckDL: + +1. **Election Safety:** At most one leader per term. + ``` + ∀ i, j ∈ Server: state[i] = Leader ∧ state[j] = Leader ∧ currentTerm[i] = currentTerm[j] → i = j + ``` + +2. **Leader Completeness:** If a log entry is committed in term T, then all leaders for terms > T have that entry. + ``` + This is the core invariant. Proved via the "Leader Completeness Property" in the Raft paper. + ``` + +3. **Log Matching:** If two entries in different logs have the same index and term, then all preceding entries are identical. + ``` + ∀ i, j ∈ Server, idx: + log[i][idx].term = log[j][idx].term → log[i][1..idx] = log[j][1..idx] + ``` + +4. **Leader Append-Only:** A leader never overwrites or deletes entries in its log. + ``` + ∀ i: state[i] = Leader → log[i] only grows (entries appended, never removed) + ``` + +5. **Term Monotonicity:** Current term never decreases. + ``` + ∀ i: currentTerm' [i] ≥ currentTerm[i] + ``` + +6. **Commit Safety:** An entry is committed only if a quorum has replicated it. + ``` + commitIndex[i] advances only when a quorum of matchIndex[i][k] ≥ N + ``` + +## KRaft Extensions Over Vanilla Raft + +KRaft is Kafka's adaptation of Raft for metadata management. Key differences: + +1. **No ZooKeeper:** KRaft replaces the ZooKeeper-based metadata quorum with a Raft-based one. +2. **Metadata log:** The replicated log contains metadata records (topic configs, partition assignments), not user data. +3. **Controller node:** KRaft has an "active controller" concept — the leader of the metadata quorum acts as the controller for the Kafka cluster. +4. **Snapshotting:** KRaft supports snapshot-based log compaction (metadata snapshots). +5. **Voters and observers:** Not all nodes are voters; some are observer nodes that replicate but don't vote. +6. **Resignation:** A leader can voluntarily step down (immediate or graceful). + +### KRaft Additional State + +| Variable | Type | Description | +|----------|------|-------------| +| `isVoter` | Server → Bool | Whether this node is a voter (vs observer) | +| `isLeader` | Server → Bool | Cached: `state[i] = Leader` | +| `highWatermark` | Nat | Similar to commitIndex; offset of last stable metadata record | + +### KRaft Additional Actions + +- **Resign:** Leader steps down gracefully +- **BecomeObserver:** Voter transitions to observer (loses vote) +- **BecomeVoter:** Observer transitions to voter (gains vote) + +## Known Verification Gaps / Interesting Targets + +1. **Election safety under network partitions:** Raft's election safety depends on the quorum overlap property. In practice, network partitions and message reordering can create edge cases. + +2. **Log truncation after leader change:** When a new leader overwrites follower logs that conflict, the truncation must be atomic. This is a known source of bugs in Raft implementations. + +3. **Stale leader problem:** After a network partition heals, a stale leader may attempt to commit entries from an old term. AdvanceCommitIndex prevents this by only committing entries from the current term. + +4. **KRaft observer consistency:** Observers don't vote but must maintain log consistency. The spec needs to ensure observers can't diverge from voters. + +5. **Graceful shutdown / resignation race:** If a leader resigns while a client request is in flight, the entry may or may not be committed. This is a liveness property, not safety, but worth verifying. + +## Producer Acknowledgement Semantics (for M1.3) + +### acks=0 (Fire and forget) +- Producer sends and doesn't wait for response +- No durability guarantee +- Invariant: **None** (messages can be lost) + +### acks=1 (Leader acknowledgment) +- Producer waits for leader to write to local log +- Durability: survives leader failure, but NOT follower failure +- Invariant: **If ack received, entry exists in leader's log** + +### acks=all (ISR acknowledgment) +- Producer waits for all in-sync replicas (ISR) to acknowledge +- Durability: survives any failure that leaves a quorum +- Invariant: **If ack received, entry is committed (quorum has replicated)** + +### Exactly-once Semantics +- Requires idempotent producer + transaction coordinator +- Idempotent producer: assigns sequence numbers; broker deduplicates +- Transaction coordinator: two-phase commit across partitions + +### Key Invariants for Producer Acks + +1. `acks=all` → no data loss on leader failover (bounded to quorum availability) +2. Idempotent producer → no duplicate messages for same PID + sequence number +3. Transaction atomicity → all partitions in a transaction commit or none do + +## Consumer Group Rebalancing (for M1.3) + +### States +- Empty → Joining → Stable → Rebalancing → Stable + +### Actions +- JoinGroup: consumer requests to join +- SyncGroup: consumer receives assignment +- Heartbeat: consumer indicates liveness +- LeaveGroup: consumer departs +- Rebalance: triggered by membership change or topic partition change + +### Invariants +1. No concurrent assignments to the same partition +2. All partitions assigned when group is stable +3. Offset monotonicity: committed offsets never decrease +4. No two consumers assigned the same partition simultaneously + +## Scope Boundary + +**In scope (M1.1):** Core Raft consensus — leader election, log replication, commit advancement, and all 6 safety invariants. + +**Deferred to M1.3:** Producer ack semantics, consumer group rebalancing, transaction coordinator. + +**Out of scope:** Full Kafka broker (produce/fetch handling), controller-specific logic, snapshot/recovery, network layer. \ No newline at end of file