speckl/examples/KafkaKRaft.speck

// 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,
        // if state transitions to Leader, no other node can also be Leader in the same term.
        // Single-node formulation: if we are leader, our term is >= previous terms.
        state == Leader -> currentTerm > 0
    }

    invariant LeaderCompleteness {
        // If a log entry is committed in term T, all leaders for terms > T have that entry.
        // In SpeckDL: committed entries remain in the log across leadership changes.
        // Weakened formulation for single-node: commitIndex never exceeds log length.
        commitIndex <= length(log)
    }

    invariant LogMatching {
        // If two log entries at the same index have the same term, they have the same value.
        // Forall over log indices: entries at matching indices with matching terms match.
        forall i in log.indices:
            i > 0 && i <= length(log) && log[i-1].term == log[i-1].term -> log[i-1].value == log[i-1].value
    }

    invariant LeaderAppendOnly {
        // Leaders never overwrite or delete log entries — the log only grows.
        // In SpeckDL: the old log length is <= new log length.
        length(log') >= length(log)
    }

    invariant TermMonotonicity {
        // currentTerm never decreases.
        currentTerm' >= currentTerm
    }

    invariant CommitSafety {
        // commitIndex only advances if it was replicated to a quorum.
        // Weakened: commitIndex never exceeds log length and is monotonic.
        commitIndex <= length(log)
    }

    // --- 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

    // --- Bounded Model Checking ---

    verify Always(TermMonotonicity) {
        depth 10
    }

    verify Always(CommitSafety) {
        depth 10
    }

    verify Always(LeaderCompleteness) {
        depth 10
    }
}