diff --git a/tla/extended_spec/MCetcdraft.cfg b/tla/extended_spec/MCetcdraft.cfg new file mode 100644 index 000000000..99c367ad5 --- /dev/null +++ b/tla/extended_spec/MCetcdraft.cfg @@ -0,0 +1,240 @@ +SPECIFICATION mc_etcdSpec + +CONSTANTS + s1 = s1 + s2 = s2 + s3 = s3 + s4 = s4 + s5 = s5 + s6 = s6 + + InitServer = {s1, s2} + Server = {s1, s2, s3, s4} + + v1 = v1 + v2 = v2 + v3 = v3 + Value = {v1, v2} + + \* Constraint limits - tuned for tractable state space + ReconfigurationLimit = 2 + MaxTermLimit = 4 + MaxTimeoutLimit = 6 + RequestLimit = 6 + + \* Fault injection limits + RestartLimit = 1 + DropLimit = 4 + DuplicateLimit = 0 + StepDownLimit = 2 + HeartbeatLimit = 8 + + \* Snapshot limits + SnapshotLimit = 2 + CompactLimit = 2 + + \* Config change limits + ConfChangeLimit = 2 + + \* ReportUnreachable limit + ReportUnreachableLimit = 1 + + \* Flow control + MaxInflightMsgs = 4 + + \* FIFO message ordering - when TRUE, enforces FIFO delivery per (source, dest) pair + MsgNoReorder = FALSE + + \* Network partition configuration + MaxPartitions = 2 \* Maximum number of partition groups (2 = split into 2 groups) + RestartDropsMessages = FALSE \* When TRUE, Restart drops all messages to/from the node + PartitionLimit = 0 \* Maximum number of partition/heal cycles + + \* DisableConfChangeValidation - when TRUE, allows proposing LeaveJoint before + \* EnterJoint is applied. This is dangerous and can violate QuorumLogInv. + \* Reference: raft.go:258-278 + DisableConfChangeValidation = FALSE + + \* Message buffer limit for state space pruning + MaxMsgBufferLimit = 12 + + \* Pending messages buffer limit for state space pruning + MaxPendingMsgLimit = 12 + + \* Action overrides + Timeout <- MCTimeout + Send <- MCSend + ClientRequest <- MCClientRequest + ClientRequestAndSend <- MCClientRequestAndSend + AddNewServer <- MCAddNewServer + DeleteServer <- MCDeleteServer + AddLearner <- MCAddLearner + Restart <- MCRestart + DropMessage <- MCDropMessage + DuplicateMessage <- MCDuplicateMessage + StepDownToFollower <- MCStepDown + Heartbeat <- MCHeartbeat + SendSnapshot <- MCSendSnapshot + CompactLog <- MCCompactLog + ChangeConf <- MCChangeConf + ChangeConfAndSend <- MCChangeConfAndSend + ManualSendSnapshot <- MCManualSendSnapshot + SendSnapshotWithCompaction <- MCSendSnapshotWithCompaction + ReportUnreachable <- MCReportUnreachable + + \* Initialization overrides + InitServerVars <- etcdInitServerVars + InitLogVars <- etcdInitLogVars + InitConfigVars <- etcdInitConfigVars + InitDurableState <- etcdInitDurableState + + \* Type constants + Nil = Nil + + ValueEntry = ValueEntry + ConfigEntry = ConfigEntry + + Follower = Follower + Candidate = Candidate + Leader = Leader + + RequestVoteRequest = RequestVoteRequest + RequestVoteResponse = RequestVoteResponse + AppendEntriesRequest = AppendEntriesRequest + AppendEntriesResponse = AppendEntriesResponse + SnapshotRequest = SnapshotRequest + SnapshotResponse = SnapshotResponse + + StateProbe = StateProbe + StateReplicate = StateReplicate + StateSnapshot = StateSnapshot + +SYMMETRY Symmetry +\* VIEW ModelView + +CHECK_DEADLOCK + FALSE + +\* Constraints +CONSTRAINT + MsgBufferConstraint + PendingMsgBufferConstraint + +\* Core Raft Safety +INVARIANTS + LogInv + MoreThanOneLeaderInv + LogMatchingInv + QuorumLogInv + MoreUpToDateCorrectInv + LeaderCompletenessInv + CommittedIsDurableInv + AppliedBoundInv + MessageDestinationValidInv + +\* Progress and Inflights +INVARIANTS + ProbeLimitInv + ProbeNetworkMessageLimitInv + ReplicatePauseInv + SnapshotInflightsInv + InflightsLogIndexInv + InflightsMatchIndexInv + InflightsMonotonicInv + ProgressStateTypeInv + InflightsInv + SnapshotPendingInv + NoPendingSnapshotInv + LeaderSelfReplicateInv + SnapshotStateInv + MatchIndexLessThanLogInv + MatchIndexNonNegativeInv + InflightsAboveMatchInv + MatchIndexLessThanNextInv + MsgAppFlowPausedConsistencyInv + ProbeOneInflightMaxInv + SnapshotNoInflightsStrictInv + NextIndexPositiveInv + NextIndexBoundInv + MatchIndexBoundInv + PendingSnapshotBoundInv + InflightsOnlyInReplicateInv + InflightsBelowNextInv + +\* Log Structure +INVARIANTS + LogOffsetMinInv + SnapshotOffsetConsistencyInv + SnapshotTermValidInv + SnapshotTermBoundInv + HistoryLogLengthInv + SnapshotCommitConsistencyInv + +\* Configuration +INVARIANTS + JointConfigNonEmptyInv + SingleConfigOutgoingEmptyInv + LearnersVotersDisjointInv + ConfigNonEmptyInv + PendingConfIndexValidInv + AppliedConfigBoundInv + +\* Message Content +INVARIANTS + SnapshotMsgIndexValidInv + SnapshotMsgTermValidInv + AppendEntriesPrevIndexNonNegInv + AppendEntriesCommitBoundInv + VoteRequestLogIndexNonNegInv + VoteRequestLogTermNonNegInv + AppendEntriesTermConsistentInv + SnapshotMsgIndexPositiveInv + ResponseTermPositiveInv + +\* Additional Safety +INVARIANTS + AllMessageTermsValid + MessageIndexValidInv + StateMachineConsistency + CommitIndexBoundInv + LogTermMonotonic + CommittedEntriesTermInv + PendingConfigBoundInv + LeaderLogLengthInv + CurrentTermAtLeastLogTerm + CandidateVotedForSelfInv + DurableStateConsistency + MessageEndpointsValidInv + LeaderDurableTermInv + +\* Term and Vote +INVARIANTS + TermPositiveInv + LeaderTermPositiveInv + CandidateTermPositiveInv + \*VotesRespondedSubsetInv + VotesGrantedSubsetInv + +\* Bug Detection +INVARIANTS + AppendEntriesPrevLogTermValidInv + SinglePendingLeaveJointInv + PendingConfIndexAutoLeaveInv + TermNeverZeroInLogInv + +\* Verdi-Raft Inspired +INVARIANTS + VotedForConsistencyInv + LeaderCurrentTermEntriesInv + RequestVoteTermBoundInv + AppendEntriesResponseTermInv + SnapshotResponseTermValidInv + +\* Temporal Properties +PROPERTIES + MonotonicCommitIndexProp + MonotonicTermProp + MonotonicMatchIndexProp + LeaderCommitCurrentTermLogsProp + LeaderAppendOnlyProp + CommittedEntriesPersistProp \ No newline at end of file diff --git a/tla/extended_spec/MCetcdraft.tla b/tla/extended_spec/MCetcdraft.tla new file mode 100644 index 000000000..7c18c2bf1 --- /dev/null +++ b/tla/extended_spec/MCetcdraft.tla @@ -0,0 +1,425 @@ +---------- MODULE MCetcdraft ---------- +\* Model checking wrapper for etcdraft + +EXTENDS etcdraft + +\* Constraint constants + +CONSTANT Value + +CONSTANT ReconfigurationLimit +ASSUME ReconfigurationLimit \in Nat + +CONSTANT MaxTermLimit +ASSUME MaxTermLimit \in Nat + +CONSTANT RequestLimit +ASSUME RequestLimit \in Nat + +\* Fault injection limits +CONSTANT RestartLimit +ASSUME RestartLimit \in Nat + +CONSTANT DropLimit +ASSUME DropLimit \in Nat + +CONSTANT DuplicateLimit +ASSUME DuplicateLimit \in Nat + +CONSTANT StepDownLimit +ASSUME StepDownLimit \in Nat + +CONSTANT HeartbeatLimit +ASSUME HeartbeatLimit \in Nat + +CONSTANT MaxTimeoutLimit +ASSUME MaxTimeoutLimit \in Nat + +CONSTANT SnapshotLimit +ASSUME SnapshotLimit \in Nat + +CONSTANT CompactLimit +ASSUME CompactLimit \in Nat + +CONSTANT ConfChangeLimit +ASSUME ConfChangeLimit \in Nat + +CONSTANT ReportUnreachableLimit +ASSUME ReportUnreachableLimit \in Nat + +CONSTANT MaxMsgBufferLimit +ASSUME MaxMsgBufferLimit \in Nat + +CONSTANT MaxPendingMsgLimit +ASSUME MaxPendingMsgLimit \in Nat + +CONSTANT PartitionLimit +ASSUME PartitionLimit \in Nat + +\* Constraint variables +VARIABLE constraintCounters + +faultVars == <> + +etcd == INSTANCE etcdraft + +\* Bootstrap initialization +BootstrapLog == + LET prevConf(y) == IF Len(y) = 0 THEN {} ELSE y[Len(y)].value.newconf + IN FoldSeq(LAMBDA x, y: Append(y, [ term |-> 1, type |-> ConfigEntry, value |-> [ newconf |-> prevConf(y) \union {x}, learners |-> {} ] ]), <<>>, SetToSeq(InitServer)) + +\* Bootstrap: InitServer members start with term 1 and config log entries +etcdInitServerVars == /\ currentTerm = [i \in Server |-> IF i \in InitServer THEN 1 ELSE 0] + /\ state = [i \in Server |-> Follower] + /\ votedFor = [i \in Server |-> Nil] + +etcdInitLogVars == /\ log = [i \in Server |-> IF i \in InitServer + THEN [offset |-> 1, entries |-> BootstrapLog, snapshotIndex |-> 0, snapshotTerm |-> 0] + ELSE [offset |-> 1, entries |-> <<>>, snapshotIndex |-> 0, snapshotTerm |-> 0]] + /\ historyLog = [i \in Server |-> IF i \in InitServer THEN BootstrapLog ELSE <<>>] + /\ commitIndex = [i \in Server |-> IF i \in InitServer THEN Cardinality(InitServer) ELSE 0] + /\ applied = [i \in Server |-> IF i \in InitServer THEN Cardinality(InitServer) ELSE 0] + +etcdInitConfigVars == /\ config = [i \in Server |-> [ jointConfig |-> IF i \in InitServer THEN <> ELSE <<{}, {}>>, learners |-> {}, autoLeave |-> FALSE]] + /\ reconfigCount = 0 \* bootstrap configs not counted + \* Bootstrap config entries are already applied (committed at Cardinality(InitServer)) + /\ appliedConfigIndex = [i \in Server |-> IF i \in InitServer THEN Cardinality(InitServer) ELSE 0] + +etcdInitDurableState == + durableState = [ i \in Server |-> [ + currentTerm |-> IF i \in InitServer THEN 1 ELSE 0, + votedFor |-> Nil, + log |-> IF i \in InitServer THEN Cardinality(InitServer) ELSE 0, + entries |-> IF i \in InitServer THEN BootstrapLog ELSE <<>>, \* Persisted log entries + snapshotIndex |-> 0, + snapshotTerm |-> 0, + snapshotHistory |-> <<>>, \* History covered by snapshot (initially empty) + commitIndex |-> IF i \in InitServer THEN Cardinality(InitServer) ELSE 0, + config |-> [ jointConfig |-> IF i \in InitServer THEN <> ELSE <<{}, {}>>, learners |-> {}, autoLeave |-> FALSE] + ]] + +\* Constrained actions + +MCAddNewServer(i, j) == + /\ reconfigCount < ReconfigurationLimit + /\ etcd!AddNewServer(i, j) + +MCDeleteServer(i, j) == + /\ reconfigCount < ReconfigurationLimit + /\ etcd!DeleteServer(i, j) + +MCAddLearner(i, j) == + /\ reconfigCount < ReconfigurationLimit + /\ etcd!AddLearner(i, j) + +\* MaxTermLimit should be >= 3 to avoid over-constraining candidate quorum +MCTimeout(i) == + /\ currentTerm[i] < MaxTermLimit + /\ constraintCounters.timeout < MaxTimeoutLimit + /\ etcd!Timeout(i) + /\ constraintCounters' = [constraintCounters EXCEPT !.timeout = @ + 1] + +MCClientRequest(i, v) == + /\ constraintCounters.request < RequestLimit + /\ etcd!ClientRequest(i, v) + /\ constraintCounters' = [constraintCounters EXCEPT !.request = @ + 1] + +MCRestart(i) == + /\ constraintCounters.restart < RestartLimit + /\ etcd!Restart(i) + /\ constraintCounters' = [constraintCounters EXCEPT !.restart = @ + 1] + +MCDropMessage(m) == + /\ constraintCounters.drop < DropLimit + /\ etcd!DropMessage(m) + /\ constraintCounters' = [constraintCounters EXCEPT !.drop = @ + 1] + +MCDuplicateMessage(m) == + /\ constraintCounters.duplicate < DuplicateLimit + /\ etcd!DuplicateMessage(m) + /\ constraintCounters' = [constraintCounters EXCEPT !.duplicate = @ + 1] + +MCHeartbeat(i, j) == + /\ constraintCounters.heartbeat < HeartbeatLimit + /\ etcd!Heartbeat(i, j) + /\ constraintCounters' = [constraintCounters EXCEPT !.heartbeat = @ + 1] + +MCStepDown(i) == + /\ constraintCounters.stepDown < StepDownLimit + /\ etcd!StepDownToFollower(i) + /\ constraintCounters' = [constraintCounters EXCEPT !.stepDown = @ + 1] + +MCSendSnapshot(i, j) == + /\ constraintCounters.snapshot < SnapshotLimit + /\ etcd!SendSnapshot(i, j) + /\ constraintCounters' = [constraintCounters EXCEPT !.snapshot = @ + 1] + +MCCompactLog(i, newStart) == + /\ constraintCounters.compact < CompactLimit + /\ etcd!CompactLog(i, newStart) + /\ constraintCounters' = [constraintCounters EXCEPT !.compact = @ + 1] + +MCManualSendSnapshot(i, j) == + /\ constraintCounters.snapshot < SnapshotLimit + /\ etcd!ManualSendSnapshot(i, j) + /\ constraintCounters' = [constraintCounters EXCEPT !.snapshot = @ + 1] + +MCSendSnapshotWithCompaction(i, j, idx) == + /\ constraintCounters.snapshot < SnapshotLimit + /\ etcd!SendSnapshotWithCompaction(i, j, idx) + /\ constraintCounters' = [constraintCounters EXCEPT !.snapshot = @ + 1] + +MCChangeConf(i) == + /\ constraintCounters.confChange < ConfChangeLimit + /\ etcd!ChangeConf(i) + /\ constraintCounters' = [constraintCounters EXCEPT !.confChange = @ + 1] + +MCChangeConfAndSend(i) == + /\ constraintCounters.confChange < ConfChangeLimit + /\ etcd!ChangeConfAndSend(i) + /\ constraintCounters' = [constraintCounters EXCEPT !.confChange = @ + 1] + +MCClientRequestAndSend(i, v) == + /\ constraintCounters.request < RequestLimit + /\ etcd!ClientRequestAndSend(i, v) + /\ constraintCounters' = [constraintCounters EXCEPT !.request = @ + 1] + +MCReportUnreachable(i, j) == + /\ constraintCounters.reportUnreachable < ReportUnreachableLimit + /\ etcd!ReportUnreachable(i, j) + /\ constraintCounters' = [constraintCounters EXCEPT !.reportUnreachable = @ + 1] + +\* Deduplicate AppendEntries: one per (source, dest, term) pair +MCSend(msg) == + /\ ~ \E n \in DOMAIN messages \union DOMAIN pendingMessages: + /\ n.mdest = msg.mdest + /\ n.msource = msg.msource + /\ n.mterm = msg.mterm + /\ n.mtype = AppendEntriesRequest + /\ msg.mtype = AppendEntriesRequest + /\ ~ \E n \in DOMAIN messages \union DOMAIN pendingMessages: + /\ n.mdest = msg.msource + /\ n.msource = msg.mdest + /\ n.mterm = msg.mterm + /\ n.mtype = AppendEntriesResponse + /\ msg.mtype = AppendEntriesRequest + /\ etcd!Send(msg) + +\* Network partition actions + +MCCreatePartition(group1) == + /\ constraintCounters.partition < PartitionLimit + /\ LET partitionAssignment == [i \in Server |-> IF i \in group1 THEN 1 ELSE 2] + IN etcd!CreatePartition(partitionAssignment) + /\ constraintCounters' = [constraintCounters EXCEPT !.partition = @ + 1] + +MCHealPartition == + /\ etcd!HealPartition + /\ UNCHANGED constraintCounters + +MCInit == + /\ etcd!Init + /\ constraintCounters = [restart |-> 0, drop |-> 0, duplicate |-> 0, stepDown |-> 0, heartbeat |-> 0, snapshot |-> 0, compact |-> 0, confChange |-> 0, reportUnreachable |-> 0, timeout |-> 0, request |-> 0, partition |-> 0] + +\* Next state relations + +MCNextAsync(i) == + \/ /\ \E j \in Server : etcd!RequestVote(i, j) + /\ UNCHANGED faultVars + \/ /\ etcd!BecomeLeader(i) + /\ UNCHANGED faultVars + \/ \E v \in Value: MCClientRequest(i, v) + \* \/ \E v \in Value: MCClientRequestAndSend(i, v) + \/ /\ etcd!AdvanceCommitIndex(i) + /\ UNCHANGED faultVars + \* Allows nextIndex below log offset to explore bug scenarios (e.g., 76f1249). + \* Correct behavior is enforced by invariants, not action constraints. + \/ /\ \E j \in Server : + /\ \E e \in nextIndex[i][j]..LastIndex(log[i])+1 : etcd!AppendEntries(i, j, <>) + /\ UNCHANGED faultVars + \/ /\ etcd!AppendEntriesToSelf(i) + /\ UNCHANGED faultVars + \/ /\ \E j \in Server : MCHeartbeat(i, j) + \/ /\ \E j \in Server : MCSendSnapshot(i, j) + \/ /\ \E j \in Server : MCManualSendSnapshot(i, j) + \/ /\ \E j \in Server : \E idx \in 1..commitIndex[i] : + /\ idx > 0 + /\ MCSendSnapshotWithCompaction(i, j, idx) + \/ /\ \E j \in Server : MCReportUnreachable(i, j) + \/ /\ etcd!ReplicateImplicitEntry(i) + /\ UNCHANGED faultVars + \* Only compact to commitIndex for most aggressive compaction scenario + \/ /\ log[i].offset < commitIndex[i] + /\ MCCompactLog(i, commitIndex[i]) + \/ MCTimeout(i) + \* Ready handled via MCReady composition + \* \/ /\ etcd!Ready(i) + \* /\ UNCHANGED faultVars + \/ MCStepDown(i) + \/ /\ \E m \in DOMAIN messages : + /\ m.mdest = i + /\ etcd!Receive(m) + /\ UNCHANGED faultVars + +\* Ready action: applies if enabled, otherwise no-op +MCReady(i) == + IF ENABLED etcd!Ready(i) + THEN /\ etcd!Ready(i) + /\ UNCHANGED faultVars + ELSE UNCHANGED <> + +\* Async action composed with Ready to reduce state space +MCNextAsyncWithReady(i) == + MCNextAsync(i) \cdot MCReady(i) + +MCNextAsyncAll == + \/ \E i \in Server : MCNextAsync(i) + \/ \E i \in Server : + /\ etcd!Ready(i) + /\ UNCHANGED faultVars + +MCNextAsyncWithReadyAll == + \E i \in Server : MCNextAsyncWithReady(i) + +MCNextCrash == \E i \in Server : MCRestart(i) + +MCNextUnreliable == + \* Only duplicate once + \/ \E m \in DOMAIN messages : + /\ messages[m] = 1 + /\ MCDuplicateMessage(m) + \* Only drop if it makes a difference + \/ \E m \in DOMAIN messages : + /\ messages[m] = 1 + /\ MCDropMessage(m) + +MCNextPartition == + \/ \E group1 \in SUBSET(Server) : + /\ group1 /= {} + /\ group1 /= Server + /\ MCCreatePartition(group1) + \/ MCHealPartition + +MCNext == + \/ MCNextAsyncAll + \/ MCNextCrash + \/ MCNextUnreliable + \/ MCNextPartition + +MCNextWithReady == + \/ MCNextAsyncWithReadyAll + \/ MCNextCrash + \/ MCNextUnreliable + \/ MCNextPartition + +\* Config changes go through ChangeConf, not direct Add/Delete +MCDynamicConfigActions == + \/ /\ \E i \in Server : MCChangeConf(i) + \* \/ /\ \E i \in Server : MCChangeConfAndSend(i) + \/ /\ \E i \in Server : etcd!ApplySimpleConfChange(i) + /\ UNCHANGED faultVars + \* ProposeLeaveJoint: AutoLeave per raft.go:745-760 + \/ /\ \E i \in Server : etcd!ProposeLeaveJoint(i) + /\ UNCHANGED faultVars + +MCNextDynamic == + \/ MCNext + \/ MCDynamicConfigActions + +MCNextDynamicWithReady == + \/ MCNextWithReady + \/ MCDynamicConfigActions + +mc_vars == <> + +mc_etcdSpec == + /\ MCInit + /\ [][MCNextDynamic]_mc_vars + +mc_etcdSpecWithReady == + /\ MCInit + /\ [][MCNextDynamicWithReady]_mc_vars + +mc_etcdSpec_no_conf_change == + /\ MCInit + /\ [][MCNext]_mc_vars + +mc_etcdSpecWithReady_no_conf_change == + /\ MCInit + /\ [][MCNextWithReady]_mc_vars + +\* Symmetry and view + +\* Enabled via cfg file; may be unsound +Symmetry == Permutations(Server) \union Permutations(Value) + +\* Excludes constraintCounters from state identity +ModelView == << view_vars >> + +\* State space pruning + +\* Limit=0 means no limit +MsgBufferConstraint == + \/ MaxMsgBufferLimit = 0 + \/ BagCardinality(messages) <= MaxMsgBufferLimit + +PendingMsgBufferConstraint == + \/ MaxPendingMsgLimit = 0 + \/ BagCardinality(pendingMessages) <= MaxPendingMsgLimit + +AllMsgBufferConstraint == + /\ MsgBufferConstraint + /\ PendingMsgBufferConstraint + +\* Monotonicity properties + +\* Excludes Restart (durableState may have older commitIndex) +MonotonicCommitIndexProp == + [][(~\E i \in Server: Restart(i)) => + \A i \in Server : commitIndex'[i] >= commitIndex[i]]_mc_vars + +\* Excludes Restart (durableState may have older term) +MonotonicTermProp == + [][(~\E i \in Server: Restart(i)) => + \A i \in Server : currentTerm'[i] >= currentTerm[i]]_mc_vars + +\* matchIndex is monotonic except during BecomeLeader, Restart, or config re-add. +\* Re-added nodes get matchIndex reset to 0 (confchange.go initProgress). +MonotonicMatchIndexProp == + [][(~ \E i \in Server: etcd!BecomeLeader(i) \/ etcd!Restart(i)) => + (\A i,j \in Server : + LET + preConfig == config[i].jointConfig[1] \cup config[i].jointConfig[2] \cup config[i].learners + postConfig == config'[i].jointConfig[1] \cup config'[i].jointConfig[2] \cup config'[i].learners + IN + (j \in preConfig /\ j \in postConfig) => matchIndex'[i][j] >= matchIndex[i][j])]_mc_vars + +\* Leader only commits entries from its current term (Raft paper Figure 8 safety) +LeaderCommitCurrentTermLogsProp == + [][ + \A i \in Server : + (state'[i] = Leader /\ commitIndex[i] /= commitIndex'[i]) => + historyLog'[i][commitIndex'[i]].term = currentTerm'[i] + ]_mc_vars + +\* Raft paper core properties + +\* Leader Append-Only (Raft Figure 3, Property 2): log only grows, existing entries unchanged +LeaderAppendOnlyProp == + [][ + \A i \in Server : + (state[i] = Leader /\ state'[i] = Leader) => + /\ LastIndex(log'[i]) >= LastIndex(log[i]) + /\ SubSeq(historyLog'[i], 1, LastIndex(log[i])) = + SubSeq(historyLog[i], 1, LastIndex(log[i])) + ]_mc_vars + +\* Committed entries persist across all transitions (excludes Restart) +CommittedEntriesPersistProp == + [][(~\E i \in Server: etcd!Restart(i)) => + \A i \in Server : + SubSeq(historyLog'[i], 1, commitIndex[i]) = + SubSeq(historyLog[i], 1, commitIndex[i]) + ]_mc_vars + +============================================================================= diff --git a/tla/extended_spec/README.md b/tla/extended_spec/README.md new file mode 100644 index 000000000..02abf5985 --- /dev/null +++ b/tla/extended_spec/README.md @@ -0,0 +1,22 @@ +# System-Level TLA+ Specification for etcd/raft + +This directory contains a system-level TLA+ specification for the etcd/raft library, complementing the protocol-level specification in `../`. The specification models implementation details including progress tracking (`StateProbe`, `StateReplicate`, `StateSnapshot`), flow control (`MsgAppFlowPaused`, Inflights), detailed snapshot handling, and joint consensus configuration changes. + +## Model Checking (Simulation) + +```bash +java -XX:+UseParallelGC \ + -cp tla2tools.jar:CommunityModules-deps.jar \ + tlc2.TLC \ + -config MCetcdraft.cfg \ + MCetcdraft.tla \ + -simulate \ + -depth 100 +``` + +## Invariants + +The specification defines **88 invariants** derived from three sources: +- **Raft Paper:** Safety properties +- **Code:** Implementation details +- **Issue/Bug:** Historical bug regression tests diff --git a/tla/extended_spec/Traceetcdraft.cfg b/tla/extended_spec/Traceetcdraft.cfg new file mode 100644 index 000000000..c30789ba0 --- /dev/null +++ b/tla/extended_spec/Traceetcdraft.cfg @@ -0,0 +1,47 @@ +CONSTANTS + Nil = "0" + + ValueEntry = "ValueEntry" + ConfigEntry = "ConfigEntry" + + Follower = "StateFollower" + Candidate = "StateCandidate" + Leader = "StateLeader" + + RequestVoteRequest = "RequestVoteRequest" + RequestVoteResponse = "RequestVoteResponse" + AppendEntriesRequest = "AppendEntriesRequest" + AppendEntriesResponse = "AppendEntriesResponse" + SnapshotRequest = "SnapshotRequest" + SnapshotResponse = "SnapshotResponse" + + StateProbe = "StateProbe" + StateReplicate = "StateReplicate" + StateSnapshot = "StateSnapshot" + + MaxInflightMsgs <- TraceMaxInflightMsgs + + \* FIFO message ordering - for trace validation, typically FALSE + MsgNoReorder = FALSE + + \* Network partition configuration - for trace validation, use defaults + MaxPartitions = 2 + RestartDropsMessages = FALSE + + \* DisableConfChangeValidation - read from trace config, default FALSE if not present + DisableConfChangeValidation <- TraceDisableConfChangeValidation + + InitServerVars <- TraceInitServerVars + InitLogVars <- TraceInitLogVars + InitConfigVars <- TraceInitConfigVars + InitLeaderVars <- TraceInitLeaderVars + + InitServer <- TraceInitServer + Server <- TraceServer + +SPECIFICATION TraceSpec + +PROPERTIES + TraceMatched + +CHECK_DEADLOCK FALSE diff --git a/tla/extended_spec/Traceetcdraft.tla b/tla/extended_spec/Traceetcdraft.tla new file mode 100644 index 000000000..13df7243d --- /dev/null +++ b/tla/extended_spec/Traceetcdraft.tla @@ -0,0 +1,667 @@ +---------------------------- MODULE Traceetcdraft -------------------------- + +EXTENDS etcdraft, Json, IOUtils, Sequences, TLC + +\* raft.pb.go enum MessageType +RaftMsgType == + "MsgApp" :> AppendEntriesRequest @@ "MsgAppResp" :> AppendEntriesResponse @@ + "MsgVote" :> RequestVoteRequest @@ "MsgVoteResp" :> RequestVoteResponse @@ + "MsgHeartbeat" :> AppendEntriesRequest @@ "MsgHeartbeatResp" :> AppendEntriesResponse @@ + "MsgSnap" :> SnapshotRequest + +RaftMsgSubtype == + "MsgHeartbeat" :> "heartbeat" @@ "MsgHeartbeatResp" :> "heartbeat" @@ + "MsgApp" :> "app" @@ "MsgAppResp" :> "app" @@ + "MsgSnap" :> "snapshot" + +RaftRole == + "StateFollower" :> Follower @@ "StateCandidate" :> Candidate @@ "StatePreCandidate" :> Candidate @@ "StateLeader" :> Leader + +------------------------------------------------------------------------------------- + +\* Trace validation has been designed for TLC running in default model-checking +\* mode, i.e., breadth-first search. +ASSUME TLCGet("config").mode = "bfs" + +JsonFile == + IF "JSON" \in DOMAIN IOEnv THEN IOEnv.JSON ELSE "./example.ndjson" + +OriginTraceLog == + \* Deserialize the System log as a sequence of records from the log file. + SelectSeq(ndJsonDeserialize(JsonFile), LAMBDA l: "event" \in DOMAIN l) + +TraceLog == + TLCEval(IF "MAX_TRACE" \in DOMAIN IOEnv THEN SubSeq(OriginTraceLog, 1, atoi(IOEnv.MAX_TRACE)) ELSE OriginTraceLog) + +TraceServer == TLCEval(FoldSeq( + LAMBDA x, y: y \cup + {x.event.nid} \cup + (IF "msg" \in DOMAIN x.event /\ "to" \in DOMAIN x.event.msg THEN {x.event.msg.to} ELSE {}) \cup + (IF x.event.name = "ChangeConf" /\ "changes" \in DOMAIN x.event.prop.cc + THEN { x.event.prop.cc.changes[k].nid : k \in 1..Len(x.event.prop.cc.changes) } + ELSE {}) \cup + (IF x.event.name = "ApplyConfChange" /\ "newconf" \in DOMAIN x.event.prop.cc THEN ToSet(x.event.prop.cc.newconf) ELSE {}), + {}, TraceLog)) + +AllChangeConfNids == TLCEval(FoldSeq( + LAMBDA x, y: y \cup + IF x.event.name = "ChangeConf" /\ "changes" \in DOMAIN x.event.prop.cc + THEN { x.event.prop.cc.changes[k].nid : k \in 1..Len(x.event.prop.cc.changes) } + ELSE {}, + {}, TraceLog)) + +BootstrapLogIndicesForServer(i) == + LET + FirstBootstrapLogIndex == SelectInSeq(TraceLog, LAMBDA x: x.event.nid = i /\ x.event.name \in {"InitState", "BecomeFollower", "ApplyConfChange"}) + FirstNonBootstrapLogIndex == SelectInSeq(TraceLog, LAMBDA x: x.event.nid = i /\ x.event.name \notin {"InitState", "BecomeFollower", "ApplyConfChange"}) + LastBootstrapLogIndexUpperBound == IF FirstNonBootstrapLogIndex = 0 THEN Len(TraceLog) ELSE FirstNonBootstrapLogIndex-1 + IN + IF FirstBootstrapLogIndex = 0 THEN {} + ELSE + { k \in FirstBootstrapLogIndex..LastBootstrapLogIndexUpperBound: TraceLog[k].event.nid = i } + +BootstrapLogIndices == UNION { BootstrapLogIndicesForServer(i): i \in Server } + +LastBootstrapLog == [ i \in Server |-> IF BootstrapLogIndicesForServer(i) = {} THEN TraceLog[1] ELSE TraceLog[Max(BootstrapLogIndicesForServer(i))] ] + +BootstrappedConfig(i) == + IF LastBootstrapLog[i].event.name = "ApplyConfChange" THEN + ToSet(LastBootstrapLog[i].event.prop.cc.newconf) + ELSE + ToSet(LastBootstrapLog[i].event.conf[1]) + +TraceInitServer == BootstrappedConfig(TraceLog[1].event.nid) +ASSUME TraceInitServer \subseteq TraceServer + +\* Extract learners from bootstrap events (InitState, BecomeFollower, BecomeCandidate) +TraceLearners == TLCEval( + LET bootstrapEvents == SelectSeq(TraceLog, LAMBDA x: + x.event.name \in {"InitState", "BecomeFollower", "BecomeCandidate"} /\ + "learners" \in DOMAIN x.event /\ x.event.learners /= <<>>) + IN IF Len(bootstrapEvents) > 0 + THEN ToSet(bootstrapEvents[1].event.learners) + ELSE {}) + +\* Use bootstrap learners if available, otherwise infer from TraceServer +ImplicitLearners == + IF TraceLearners /= {} + THEN TraceLearners + ELSE TraceServer \ (TraceInitServer \cup AllChangeConfNids) + +\* Extract MaxInflightMsgs from trace config (default: 256) +TraceMaxInflightMsgs == TLCEval( + LET configLines == SelectSeq(ndJsonDeserialize(JsonFile), LAMBDA x: "tag" \in DOMAIN x /\ x.tag = "config") + IN IF Len(configLines) > 0 /\ "config" \in DOMAIN configLines[1] /\ "MaxInflightMsgs" \in DOMAIN configLines[1].config + THEN configLines[1].config.MaxInflightMsgs + ELSE 256 +) + +\* Extract DisableConfChangeValidation from trace config (default: FALSE) +TraceDisableConfChangeValidation == TLCEval( + LET configLines == SelectSeq(ndJsonDeserialize(JsonFile), LAMBDA x: "tag" \in DOMAIN x /\ x.tag = "config") + IN IF Len(configLines) > 0 /\ "config" \in DOMAIN configLines[1] /\ "DisableConfChangeValidation" \in DOMAIN configLines[1].config + THEN configLines[1].config.DisableConfChangeValidation + ELSE FALSE +) + +TraceInitServerVars == + /\ currentTerm = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} THEN 0 ELSE LastBootstrapLog[i].event.state.term] + /\ state = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} THEN Follower ELSE LastBootstrapLog[i].event.role] + /\ votedFor = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} THEN "0" ELSE LastBootstrapLog[i].event.state.vote] + +TraceInitLogVars == + /\ log = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} + THEN [offset |-> 1, entries |-> <<>>, snapshotIndex |-> 0, snapshotTerm |-> 0] + ELSE [offset |-> 1, + entries |-> [j \in 1..LastBootstrapLog[i].event.log |-> [ term |-> 1, type |-> "ConfigEntry", value |-> [newconf |-> BootstrappedConfig(i), learners |-> ImplicitLearners]]], + snapshotIndex |-> LastBootstrapLog[i].event.state.snapshotIndex, + snapshotTerm |-> LastBootstrapLog[i].event.state.snapshotTerm + ] + ] + /\ historyLog = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} THEN <<>> ELSE [j \in 1..LastBootstrapLog[i].event.log |-> [ term |-> 1, type |-> "ConfigEntry", value |-> [newconf |-> BootstrappedConfig(i), learners |-> ImplicitLearners]]]] + /\ commitIndex = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} THEN 0 ELSE LastBootstrapLog[i].event.state.commit] + /\ applied = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} THEN 0 ELSE LastBootstrapLog[i].event.state.applied] + +TraceInitConfigVars == + /\ config = [i \in Server |-> [ jointConfig |-> <>, learners |-> ImplicitLearners, autoLeave |-> FALSE] ] + /\ reconfigCount = 0 + \* Bootstrap config entries are already applied (committed at initial commitIndex) + /\ appliedConfigIndex = [i \in Server |-> IF BootstrapLogIndicesForServer(i)={} THEN 0 ELSE LastBootstrapLog[i].event.state.commit] + +TraceInitLeaderVars == + /\ matchIndex = [i \in Server |-> [j \in Server |-> 0]] + \* Initialize pendingConfChangeIndex from trace if available + /\ pendingConfChangeIndex = [i \in Server |-> + IF BootstrapLogIndicesForServer(i) = {} + THEN 0 + ELSE IF "pendingConfIndex" \in DOMAIN LastBootstrapLog[i].event.state + THEN LastBootstrapLog[i].event.state.pendingConfIndex + ELSE 0] + +------------------------------------------------------------------------------------- +ConfFromLog(l) == << ToSet(l.event.conf[1]), ToSet(l.event.conf[2]) >> + +OneMoreMessage(msg) == + \/ msg \notin DOMAIN pendingMessages /\ msg \in DOMAIN pendingMessages' /\ pendingMessages'[msg] = 1 + \/ msg \in DOMAIN pendingMessages /\ pendingMessages'[msg] = pendingMessages[msg] + 1 + +OneLessMessage(msg) == + \/ msg \in DOMAIN messages /\ messages[msg] = 1 /\ msg \notin DOMAIN messages' + \/ msg \in DOMAIN messages /\ messages'[msg] = messages[msg] - 1 + +------------------------------------------------------------------------------------- + +VARIABLE l +logline == TraceLog[l] +VARIABLE pl + + +TraceInit == + /\ l = 1 + /\ pl = 0 + /\ logline = TraceLog[l] + /\ Init + +StepToNextTrace == + /\ l' = l+1 + /\ pl' = l + /\ l % Max({1, Len(TraceLog) \div 100}) = 0 => PrintT(<< "Progress %:", (l * 100) \div Len(TraceLog)>>) + /\ l' > Len(TraceLog) => PrintT(<< "Progress %:", 100>>) + +StepToNextTraceIfMessageIsProcessed(msg) == + IF OneLessMessage(msg) + THEN StepToNextTrace + ELSE + /\ pl' = l + /\ UNCHANGED <> + +------------------------------------------------------------------------------------- + +LoglineIsEvent(e) == + /\ l <= Len(TraceLog) + /\ logline.event.name = e + +LoglineIsEvents(e) == + /\ l <= Len(TraceLog) + /\ logline.event.name \in e + +LoglineIsMessageEvent(e, i, j) == + /\ LoglineIsEvent(e) + /\ logline.event.msg.from = i + /\ logline.event.msg.to = j + +LoglineIsNodeEvent(e, i) == + /\ LoglineIsEvent(e) + /\ logline.event.nid = i + +LoglineIsAppendEntriesRequest(m) == + /\ "msg" \in DOMAIN logline.event + /\ m.mtype = AppendEntriesRequest + /\ m.mtype = RaftMsgType[logline.event.msg.type] + /\ m.msubtype = RaftMsgSubtype[logline.event.msg.type] + /\ m.mdest = logline.event.msg.to + /\ m.msource = logline.event.msg.from + /\ m.mterm = logline.event.msg.term + /\ m.msubtype /= "snapshot" => m.mcommitIndex = logline.event.msg.commit + /\ m.msubtype /= "heartbeat" => /\ m.mprevLogTerm = logline.event.msg.logTerm + /\ m.mprevLogIndex = logline.event.msg.index + /\ Len(m.mentries) = logline.event.msg.entries + +LoglineIsSnapshotRequest(m) == + /\ "msg" \in DOMAIN logline.event + /\ m.mtype = SnapshotRequest + /\ m.mtype = RaftMsgType[logline.event.msg.type] + /\ m.mdest = logline.event.msg.to + /\ m.msource = logline.event.msg.from + /\ m.mterm = logline.event.msg.term + /\ m.msnapshotIndex = logline.event.msg.index + /\ m.msnapshotTerm = logline.event.msg.logTerm + +LoglineIsAppendEntriesResponse(m) == + /\ "msg" \in DOMAIN logline.event + /\ m.mtype = AppendEntriesResponse + /\ m.mtype = RaftMsgType[logline.event.msg.type] + /\ \/ m.msubtype = RaftMsgSubtype[logline.event.msg.type] + \/ /\ logline.event.msg.type = "MsgAppResp" + /\ m.msubtype = "snapshot" + /\ m.mdest = logline.event.msg.to + /\ m.msource = logline.event.msg.from + /\ m.mterm = logline.event.msg.term + /\ m.msuccess = ~logline.event.msg.reject + \* For success: mmatchIndex = index, mrejectHint = 0, mlogTerm = 0 + \* For reject: mmatchIndex = rejected index, mrejectHint/mlogTerm from trace + /\ (~logline.event.msg.reject /\ m.msubtype /= "heartbeat") => m.mmatchIndex = logline.event.msg.index + /\ logline.event.msg.reject => /\ m.mmatchIndex = logline.event.msg.index + /\ m.mrejectHint = logline.event.msg.rejectHint + /\ m.mlogTerm = logline.event.msg.logTerm + +LoglineIsRequestVoteRequest(m) == + /\ "msg" \in DOMAIN logline.event + /\ m.mtype = RequestVoteRequest + /\ m.mtype = RaftMsgType[logline.event.msg.type] + /\ m.mdest = logline.event.msg.to + /\ m.msource = logline.event.msg.from + /\ m.mterm = logline.event.msg.term + /\ m.mlastLogIndex = logline.event.msg.index + /\ m.mlastLogTerm = logline.event.msg.logTerm + +LoglineIsRequestVoteResponse(m) == + /\ "msg" \in DOMAIN logline.event + /\ m.mtype = RequestVoteResponse + /\ m.mtype = RaftMsgType[logline.event.msg.type] + /\ m.mdest = logline.event.msg.to + /\ m.msource = logline.event.msg.from + /\ m.mterm = logline.event.msg.term + /\ m.mvoteGranted = ~logline.event.msg.reject + +\* Helper to access log entry at specific logical index from a log record +LogEntryAt(xlog, index) == + xlog.entries[index - xlog.offset + 1] + +ValidatePreStates(i) == + pl = l - 1 => + /\ currentTerm[i] = logline.event.state.term + /\ state[i] = RaftRole[logline.event.role] + /\ commitIndex[i] = logline.event.state.commit + \* Note: applied not validated - updated asynchronously by application layer + /\ votedFor[i] = logline.event.state.vote + +ValidatePostStates(i) == + /\ currentTerm'[i] = logline.event.state.term + /\ state'[i] = logline.event.role + /\ votedFor'[i] = logline.event.state.vote + /\ LastIndex(log'[i]) = logline.event.log + /\ commitIndex'[i] = logline.event.state.commit + /\ config'[i].jointConfig = ConfFromLog(logline) + /\ log'[i].snapshotIndex = logline.event.state.snapshotIndex + /\ log'[i].snapshotTerm = logline.event.state.snapshotTerm + \* Validate applied if present in trace + /\ "applied" \in DOMAIN logline.event.state => + applied'[i] = logline.event.state.applied + \* Validate config.learners if present in trace + /\ "learners" \in DOMAIN logline.event => + config'[i].learners = ToSet(logline.event.learners) + +------------------------------------------------------------------------------------- +\* Progress-specific validation helpers + +ValidateProgressState(i, j) == + \/ /\ "prop" \notin DOMAIN logline.event + /\ TRUE + \/ /\ "prop" \in DOMAIN logline.event + /\ "state" \in DOMAIN logline.event.prop + /\ progressState[i][j] = logline.event.prop.state + /\ "match" \in DOMAIN logline.event.prop => + matchIndex[i][j] = logline.event.prop.match + /\ "next" \in DOMAIN logline.event.prop => + nextIndex[i][j] = logline.event.prop.next + /\ "paused" \in DOMAIN logline.event.prop => + msgAppFlowPaused[i][j] = logline.event.prop.paused + /\ "inflights_count" \in DOMAIN logline.event.prop => + Cardinality(inflights[i][j]) = logline.event.prop.inflights_count + /\ (progressState[i][j] = StateSnapshot /\ "pending_snapshot" \in DOMAIN logline.event.prop) => + pendingSnapshot[i][j] = logline.event.prop.pending_snapshot + +------------------------------------------------------------------------------------- + +ValidateAfterRequestVote(i, j) == + /\ ValidatePostStates(i) + /\ \E m \in DOMAIN pendingMessages': + /\ \/ LoglineIsRequestVoteRequest(m) + \/ /\ LoglineIsRequestVoteResponse(m) + /\ m.msource = m.mdest + /\ OneMoreMessage(m) + +RequestVoteIfLogged(i, j) == + /\ \/ LoglineIsMessageEvent("SendRequestVoteRequest", i, j) + \/ /\ LoglineIsMessageEvent("SendRequestVoteResponse", i, j) + /\ i = j + /\ RequestVote(i, j) + /\ ValidateAfterRequestVote(i, j) + +ValidateAfterBecomeLeader(i) == + /\ ValidatePostStates(i) + /\ logline.event.role = "StateLeader" + /\ state'[i] = Leader + /\ \A j \in Server: j /= i => progressState'[i][j] = StateProbe + +BecomeLeaderIfLogged(i) == + /\ LoglineIsNodeEvent("BecomeLeader", i) + /\ BecomeLeader(i) + /\ ValidateAfterBecomeLeader(i) + +ClientRequestIfLogged(i, v) == + /\ LoglineIsNodeEvent("Replicate", i) + /\ ClientRequest(i, v) + +ValidateAfterAdvanceCommitIndex(i) == + /\ ValidatePostStates(i) + /\ logline.event.role = "StateLeader" + /\ state[i] = Leader + +AdvanceCommitIndexIfLogged(i) == + /\ LoglineIsNodeEvent("Commit", i) + /\ state[i] = Leader \* All Commit events in trace are from Leader + /\ AdvanceCommitIndex(i) + /\ ValidateAfterAdvanceCommitIndex(i) + +ValidateProgressStatePrimed(i, j) == + \/ /\ "prop" \notin DOMAIN logline.event + /\ TRUE + \/ /\ "prop" \in DOMAIN logline.event + /\ "state" \in DOMAIN logline.event.prop + /\ progressState'[i][j] = logline.event.prop.state + /\ "match" \in DOMAIN logline.event.prop => + matchIndex'[i][j] = logline.event.prop.match + /\ "next" \in DOMAIN logline.event.prop => + nextIndex'[i][j] = logline.event.prop.next + /\ "paused" \in DOMAIN logline.event.prop => + msgAppFlowPaused'[i][j] = logline.event.prop.paused + /\ "inflights_count" \in DOMAIN logline.event.prop => + Cardinality(inflights'[i][j]) = logline.event.prop.inflights_count + /\ (progressState'[i][j] = StateSnapshot /\ "pending_snapshot" \in DOMAIN logline.event.prop) => + pendingSnapshot'[i][j] = logline.event.prop.pending_snapshot + +ValidateAfterAppendEntries(i, j) == + /\ ValidatePostStates(i) + /\ \E msg \in DOMAIN pendingMessages': + /\ LoglineIsAppendEntriesRequest(msg) + /\ OneMoreMessage(msg) + /\ ValidateProgressState(i, j) + +ValidateAfterHeartbeat(i, j) == + /\ ValidatePostStates(i) + /\ \E msg \in DOMAIN pendingMessages': + /\ LoglineIsAppendEntriesRequest(msg) + /\ OneMoreMessage(msg) + /\ ValidateProgressState(i, j) + +ValidateAfterAppendEntriesToSelf(i) == + /\ ValidatePostStates(i) + /\ \E msg \in DOMAIN pendingMessages': + /\ LoglineIsAppendEntriesResponse(msg) + /\ msg.msource = msg.mdest + /\ OneMoreMessage(msg) + +ValidateAfterSnapshot(i, j) == + /\ ValidatePostStates(i) + /\ \E msg \in DOMAIN pendingMessages': + /\ LoglineIsSnapshotRequest(msg) + /\ OneMoreMessage(msg) + /\ ValidateProgressStatePrimed(i, j) + +AppendEntriesIfLogged(i, j, range) == + /\ LoglineIsMessageEvent("SendAppendEntriesRequest", i, j) + /\ logline.event.msg.type = "MsgApp" + /\ range[1] = logline.event.msg.index + 1 + /\ range[2] = range[1] + logline.event.msg.entries + /\ AppendEntries(i, j, range) + /\ ValidateAfterAppendEntries(i, j) + +HeartbeatIfLogged(i, j) == + /\ LoglineIsMessageEvent("SendAppendEntriesRequest", i, j) + /\ logline.event.msg.type = "MsgHeartbeat" + /\ Heartbeat(i, j) + /\ ValidateAfterAppendEntries(i, j) + +\* SendSnapshot - uses SendSnapshotWithCompaction +SendSnapshotIfLogged(i, j, index) == + /\ LoglineIsMessageEvent("SendAppendEntriesRequest", i, j) + /\ logline.event.msg.type = "MsgSnap" + /\ index = logline.event.msg.index + /\ SendSnapshotWithCompaction(i, j, index) \* Call action from etcdraft.tla + /\ ValidateAfterSnapshot(i, j) + /\ progressState'[i][j] = StateSnapshot + +\* ManualSendSnapshot bypasses normal raft send path, does NOT modify progress state +ManualSendSnapshotIfLogged(i, j) == + /\ LoglineIsMessageEvent("ManualSendSnapshot", i, j) + /\ logline.event.msg.type = "MsgSnap" + /\ ManualSendSnapshot(i, j) \* Call action from etcdraft.tla + + +\* Implicit replication with self-directed MsgAppResp +ImplicitReplicateAndSend(i) == + ReplicateImplicitEntry(i) \* Call action from etcdraft.tla + +AppendEntriesToSelfIfLogged(i) == + /\ LoglineIsMessageEvent("SendAppendEntriesResponse", i, i) + /\ IF LastIndex(log[i]) < logline.event.log + THEN ImplicitReplicateAndSend(i) /\ ValidateAfterAppendEntriesToSelf(i) + ELSE AppendEntriesToSelf(i) /\ ValidateAfterAppendEntriesToSelf(i) + +ReceiveMessageTraceNames == { "ReceiveAppendEntriesRequest", "ReceiveAppendEntriesResponse", "ReceiveRequestVoteRequest", "ReceiveRequestVoteResponse", "ReceiveSnapshot" } +\* perform Receive transition if logline indicates so +LoglineIsReceivedMessage(m) == + \/ /\ LoglineIsEvent("ReceiveAppendEntriesRequest") + /\ \/ LoglineIsAppendEntriesRequest(m) + \/ LoglineIsSnapshotRequest(m) + \/ /\ LoglineIsEvent("ReceiveAppendEntriesResponse") + /\ LoglineIsAppendEntriesResponse(m) + \/ /\ LoglineIsEvent("ReceiveRequestVoteRequest") + /\ LoglineIsRequestVoteRequest(m) + \/ /\ LoglineIsEvent("ReceiveRequestVoteResponse") + /\ LoglineIsRequestVoteResponse(m) + \* ReceiveSnapshot is actually covered by ReceiveAppendEntriesRequest in state_trace.go, + \* but if trace uses ReceiveSnapshot explicit tag, we handle it here. + \/ /\ LoglineIsEvent("ReceiveSnapshot") + /\ LoglineIsSnapshotRequest(m) + +ReceiveIfLogged(m) == + /\ LoglineIsReceivedMessage(m) + /\ ValidatePreStates(m.mdest) + /\ Receive(m) + +\* perform Timeout transition if logline indicates so +ValidateAfterTimeout(i) == + /\ logline.event.role = "StateCandidate" + /\ logline.event.nid = i + /\ state'[i] = Candidate + /\ currentTerm'[i] = logline.event.state.term + +TimeoutIfLogged(i) == + /\ LoglineIsNodeEvent("BecomeCandidate", i) + /\ Timeout(i) + /\ ValidateAfterTimeout(i) + + +ChangeConfIfLogged(i) == + /\ LoglineIsNodeEvent("ChangeConf", i) + /\ ValidatePreStates(i) + /\ LET changes == logline.event.prop.cc.changes + initialConf == [voters |-> GetConfig(i), learners |-> GetLearners(i)] + finalConf == FoldSeq(ApplyChange, initialConf, changes) + enterJoint == logline.event.prop.enterJoint + IN + \* Use ChangeConf's IF-branch directly: BecomeLeader sets pendingConfChangeIndex = lastIndex, + \* so ChangeConf(i) would take the ELSE branch, but trace proves a real config change happened. + /\ state[i] = Leader + /\ ~IsJointConfig(i) + /\ Replicate(i, [newconf |-> finalConf.voters, learners |-> finalConf.learners, + enterJoint |-> enterJoint, oldconf |-> GetConfig(i)], ConfigEntry) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i] = LastIndex(log'[i])] + \* Validate the created entry matches trace expectations + /\ LogEntryAt(log'[i], LastIndex(log'[i])).value.newconf = finalConf.voters + /\ LogEntryAt(log'[i], LastIndex(log'[i])).value.learners = finalConf.learners + /\ LogEntryAt(log'[i], LastIndex(log'[i])).value.enterJoint = enterJoint + /\ LogEntryAt(log'[i], LastIndex(log'[i])).value.oldconf = GetConfig(i) + /\ UNCHANGED <> + +ApplySimpleConfChangeIfLogged(i) == + /\ LoglineIsNodeEvent("ApplyConfChange", i) + /\ ~IsJointConfig(i) \* If we're in joint, we should be using LeaveJointIfLogged instead + \* Exclude LeaveJoint events (they have leaveJoint flag) + /\ ~("leaveJoint" \in DOMAIN logline.event.prop.cc /\ logline.event.prop.cc.leaveJoint = TRUE) + /\ ApplySimpleConfChange(i) + +\* Leave joint consensus via ApplySimpleConfChange +LeaveJointIfLogged(i) == + /\ LoglineIsNodeEvent("ApplyConfChange", i) + /\ "newconf" \in DOMAIN logline.event.prop.cc + /\ \/ IsJointConfig(i) \* We're currently in joint consensus + \/ ("leaveJoint" \in DOMAIN logline.event.prop.cc /\ logline.event.prop.cc.leaveJoint = TRUE) + /\ ApplySimpleConfChange(i) \* Use ApplySimpleConfChange to update appliedConfigIndex + +\* Apply configuration from snapshot +ApplySnapshotConfChangeIfLogged(i) == + /\ LoglineIsNodeEvent("ApplyConfChange", i) + /\ "newconf" \in DOMAIN logline.event.prop.cc + /\ log[i].offset > commitIndex[i] \* Indicates snapshot was applied + /\ LET newVoters == ToSet(logline.event.prop.cc.newconf) + IN ApplySnapshotConfChange(i, newVoters) \* Call action from etcdraft.tla + +ReadyIfLogged(i) == + /\ LoglineIsNodeEvent("Ready", i) + /\ Ready(i) + +RestartIfLogged(i) == + /\ LoglineIsNodeEvent("InitState", i) + /\ Restart(i) + /\ ValidatePostStates(i) + +LoglineIsBecomeFollowerInUpdateTermOrReturnToFollower == + /\ LoglineIsEvent("BecomeFollower") + /\ LET + k == SelectLastInSubSeq(TraceLog, 1, l-1, LAMBDA x: x.event.nid = logline.event.nid) + IN + /\ k > 0 + /\ \/ /\ TraceLog[k].event.name \in ReceiveMessageTraceNames + /\ TraceLog[k].event.state.term < TraceLog[k].event.msg.term + /\ TraceLog[k].event.msg.term = logline.event.state.term + \/ /\ TraceLog[k].event.name = "ReceiveAppendEntriesRequest" + /\ TraceLog[k].event.state.term = TraceLog[k].event.msg.term + /\ TraceLog[k].event.msg.term = logline.event.state.term + /\ TraceLog[k].event.role = Candidate + +StepDownToFollowerIfLogged(i) == + /\ LoglineIsNodeEvent("BecomeFollower", i) + /\ \lnot LoglineIsBecomeFollowerInUpdateTermOrReturnToFollower + /\ StepDownToFollower(i) + /\ ValidatePostStates(i) + +\* Handle ReportUnreachable event +ReportUnreachableIfLogged(i, j) == + /\ LoglineIsNodeEvent("ReportUnreachable", i) + /\ "prop" \in DOMAIN logline.event + /\ "target" \in DOMAIN logline.event.prop + /\ j = logline.event.prop.target + /\ ReportUnreachable(i, j) + \* Use Primed version since trace records state AFTER the transition + /\ ValidateProgressStatePrimed(i, j) + +\* Handle ReportSnapshotStatus event +ReportSnapshotStatusIfLogged(i, j) == + /\ LoglineIsNodeEvent("ReportSnapshotStatus", i) + /\ "prop" \in DOMAIN logline.event + /\ "target" \in DOMAIN logline.event.prop + /\ "success" \in DOMAIN logline.event.prop + /\ j = logline.event.prop.target + /\ ReportSnapshotStatus(i, j, logline.event.prop.success) + \* Use Primed version since trace records state AFTER the transition + /\ ValidateProgressStatePrimed(i, j) + +\* skip unused logs +SkipUnusedLogline == + /\ \/ /\ LoglineIsEvent("SendRequestVoteResponse") + /\ logline.event.msg.from # logline.event.msg.to + \/ /\ LoglineIsEvent("SendAppendEntriesResponse") + /\ logline.event.msg.from # logline.event.msg.to \* Non-self response already handled by HandleAppendEntriesRequest + \/ LoglineIsBecomeFollowerInUpdateTermOrReturnToFollower + \/ LoglineIsEvent("ReduceNextIndex") \* shall not be necessary when this is removed from raft + /\ UNCHANGED <> + /\ StepToNextTrace + +\* Handle ApplyEntries event +ApplyEntriesIfLogged(i) == + /\ LoglineIsNodeEvent("ApplyEntries", i) + /\ "prop" \in DOMAIN logline.event + /\ "applied" \in DOMAIN logline.event.prop + /\ LET newApplied == logline.event.prop.applied + IN ApplyEntries(i, newApplied) + /\ ValidatePostStates(i) + +CompactLogIfLogged(i) == + /\ LoglineIsNodeEvent("CompactLog", i) + /\ LET newStart == logline.event.prop.compactIndex + 1 \* compactIndex is snapshotIndex, newStart = snapshotIndex + 1 + IN CompactLog(i, newStart) + /\ ValidatePostStates(i) + +TraceNextNonReceiveActions == + /\ \/ /\ LoglineIsEvents({"SendRequestVoteRequest", "SendRequestVoteResponse"}) + /\ \E i,j \in Server : RequestVoteIfLogged(i, j) + \/ /\ LoglineIsEvent("BecomeLeader") + /\ \E i \in Server : BecomeLeaderIfLogged(i) + \/ /\ LoglineIsEvent("Replicate") + /\ \E i \in Server : ClientRequestIfLogged(i, 0) + \/ /\ LoglineIsEvent("Commit") + /\ \E i \in Server : AdvanceCommitIndexIfLogged(i) + \/ /\ LoglineIsEvent("SendAppendEntriesRequest") /\ logline.event.msg.type = "MsgApp" + /\ \E i,j \in Server : \E b,e \in matchIndex[i][j]+1..LastIndex(log[i])+1 : AppendEntriesIfLogged(i, j, <>) + \/ /\ LoglineIsEvent("SendAppendEntriesResponse") + /\ \E i \in Server : AppendEntriesToSelfIfLogged(i) + \/ /\ LoglineIsEvent("SendAppendEntriesRequest") + /\ \E i,j \in Server : HeartbeatIfLogged(i, j) /\ logline.event.msg.type = "MsgHeartbeat" + \/ /\ LoglineIsEvent("SendAppendEntriesRequest") /\ logline.event.msg.type = "MsgSnap" + \* Pass msg.index as snapshot index + /\ \E i,j \in Server : SendSnapshotIfLogged(i, j, logline.event.msg.index) + \/ /\ LoglineIsEvent("ManualSendSnapshot") + /\ \E i,j \in Server : ManualSendSnapshotIfLogged(i, j) + \/ /\ LoglineIsEvent("BecomeCandidate") + /\ \E i \in Server : TimeoutIfLogged(i) + \/ /\ LoglineIsEvent("ChangeConf") + /\ \E i \in Server: ChangeConfIfLogged(i) + \/ /\ LoglineIsEvent("ApplyConfChange") + /\ \E i \in Server: \/ ApplySimpleConfChangeIfLogged(i) + \/ ApplySnapshotConfChangeIfLogged(i) + \/ LeaveJointIfLogged(i) + \/ /\ LoglineIsEvent("Ready") + /\ \E i \in Server: ReadyIfLogged(i) + \/ /\ LoglineIsEvent("InitState") + /\ \E i \in Server: RestartIfLogged(i) + \/ /\ LoglineIsEvent("BecomeFollower") + /\ \E i \in Server: StepDownToFollowerIfLogged(i) + \/ /\ LoglineIsEvent("ReportUnreachable") + /\ \E i,j \in Server: ReportUnreachableIfLogged(i, j) + \/ /\ LoglineIsEvent("ReportSnapshotStatus") + /\ \E i,j \in Server: ReportSnapshotStatusIfLogged(i, j) + \/ /\ LoglineIsEvent("ApplyEntries") + /\ \E i \in Server: ApplyEntriesIfLogged(i) + \/ /\ LoglineIsEvent("CompactLog") + /\ \E i \in Server: CompactLogIfLogged(i) + \/ SkipUnusedLogline + /\ StepToNextTrace + +TraceNextReceiveActions == + /\ LoglineIsEvents(ReceiveMessageTraceNames) + /\ \E m \in DOMAIN messages : + /\ ReceiveIfLogged(m) + /\ StepToNextTraceIfMessageIsProcessed(m) + +TraceNext == + \/ /\ l \in BootstrapLogIndices + /\ UNCHANGED <> + /\ StepToNextTrace + \/ /\ l \notin BootstrapLogIndices + /\ \/ TraceNextNonReceiveActions + \/ TraceNextReceiveActions + +TraceSpec == + TraceInit /\ [][TraceNext]_<> + +------------------------------------------------------------------------------------- + +TraceView == + <> + +------------------------------------------------------------------------------------- + +\* The property TraceMatched below will be violated if TLC runs with more than a single worker. +ASSUME TLCGet("config").worker = 1 + +TraceMatched == + [](l <= Len(TraceLog) => [](TLCGet("queue") = 1 \/ l > Len(TraceLog))) + +etcd == INSTANCE etcdraft +etcdSpec == etcd!Init /\ [][etcd!NextDynamic]_etcd!vars + +================================================================================== diff --git a/tla/extended_spec/etcdraft.cfg b/tla/extended_spec/etcdraft.cfg new file mode 100644 index 000000000..6a320bf16 --- /dev/null +++ b/tla/extended_spec/etcdraft.cfg @@ -0,0 +1,53 @@ +SPECIFICATION Spec + +CONSTANTS + s1 = s1 + s2 = s2 + s3 = s3 + + InitServer = {s1, s2, s3} + Server = {s1, s2, s3} + + Nil = Nil + + ValueEntry = ValueEntry + ConfigEntry = ConfigEntry + + Follower = Follower + Candidate = Candidate + Leader = Leader + + RequestVoteRequest = RequestVoteRequest + RequestVoteResponse = RequestVoteResponse + AppendEntriesRequest = AppendEntriesRequest + AppendEntriesResponse = AppendEntriesResponse + SnapshotRequest = SnapshotRequest + SnapshotResponse = SnapshotResponse + + \* Progress state constants + StateProbe = StateProbe + StateReplicate = StateReplicate + StateSnapshot = StateSnapshot + + \* Flow control configuration + MaxInflightMsgs = 2 + + \* FIFO message ordering + MsgNoReorder = FALSE + + \* Network partition configuration + MaxPartitions = 2 + RestartDropsMessages = FALSE + + \* Configuration change validation + DisableConfChangeValidation = FALSE + +\* Core Raft safety invariants +INVARIANTS + LogInv + MoreThanOneLeaderInv + LogMatchingInv + QuorumLogInv + MoreUpToDateCorrectInv + LeaderCompletenessInv + diff --git a/tla/extended_spec/etcdraft.tla b/tla/extended_spec/etcdraft.tla new file mode 100644 index 000000000..603cc1525 --- /dev/null +++ b/tla/extended_spec/etcdraft.tla @@ -0,0 +1,2406 @@ +-------------------------- MODULE etcdraft -------------------------- +EXTENDS Naturals, Integers, Bags, FiniteSets, Sequences, SequencesExt, FiniteSetsExt, BagsExt, TLC + +\* The initial and global set of server IDs. +CONSTANTS InitServer, Server + +\* Log metadata to distinguish values from configuration changes. +CONSTANT ValueEntry, ConfigEntry + +\* Server states. +CONSTANTS + \* @type: Str; + Follower, + \* @type: Str; + Candidate, + \* @type: Str; + Leader + +\* A reserved value. +CONSTANTS + \* @type: Int; + Nil + +\* Message types: +CONSTANTS + \* @type: Str; + RequestVoteRequest, + \* @type: Str; + RequestVoteResponse, + \* @type: Str; + AppendEntriesRequest, + \* @type: Str; + AppendEntriesResponse, + \* @type: Str; + SnapshotRequest, + \* @type: Str; + SnapshotResponse + +\* Progress state constants +\* Reference: tracker/state.go:20-33 +CONSTANTS + \* @type: Str; + StateProbe, \* Probe state: don't know follower's last index + \* @type: Str; + StateReplicate, \* Replicate state: normal fast replication + \* @type: Str; + StateSnapshot \* Snapshot state: need to send snapshot + +\* Flow control configuration constants +\* Reference: raft.go:205-210, Config.MaxInflightMsgs +CONSTANT + \* @type: Int; + MaxInflightMsgs \* Max inflight messages (per Progress) + +ASSUME MaxInflightMsgs \in Nat /\ MaxInflightMsgs > 0 + +\* When TRUE, enforce FIFO message ordering via sequence numbers +CONSTANT MsgNoReorder +ASSUME MsgNoReorder \in BOOLEAN + +\* Maximum number of network partitions (2 = split into 2 groups) +CONSTANT MaxPartitions +ASSUME MaxPartitions \in Nat /\ MaxPartitions >= 1 + +\* When TRUE, Restart drops all messages to/from the restarting node +CONSTANT RestartDropsMessages +ASSUME RestartDropsMessages \in BOOLEAN + +\* Turns off propose-time config change verification. +\* Reference: raft.go:258-278 +\* WARNING: can violate QuorumLogInv if new config members lack committed entries. +CONSTANT DisableConfChangeValidation +ASSUME DisableConfChangeValidation \in BOOLEAN + +---- +\* A bag of records representing requests and responses sent between servers. +VARIABLE + \* @typeAlias: ENTRY = [term: Int, value: Int]; + \* @typeAlias: LOGT = Seq(ENTRY); + \* @typeAlias: RVREQT = [mtype: Str, mterm: Int, mlastLogTerm: Int, mlastLogIndex: Int, msource: Int, mdest: Int]; + \* @typeAlias: RVRESPT = [mtype: Str, mterm: Int, mvoteGranted: Bool, msource: Int, mdest: Int ]; + \* @typeAlias: AEREQT = [mtype: Str, mterm: Int, mprevLogIndex: Int, mprevLogTerm: Int, mentries: LOGT, mcommitIndex: Int, msource: Int, mdest: Int ]; + \* @typeAlias: AERESPT = [mtype: Str, mterm: Int, msuccess: Bool, mmatchIndex: Int, mrejectHint: Int, mlogTerm: Int, msource: Int, mdest: Int ]; + \* @typeAlias: MSG = [ wrapped: Bool, mtype: Str, mterm: Int, msource: Int, mdest: Int, RVReq: RVREQT, RVResp: RVRESPT, AEReq: AEREQT, AEResp: AERESPT ]; + \* @type: MSG -> Int; + messages +VARIABLE + pendingMessages + +\* FIFO sequence counter (used when MsgNoReorder = TRUE) +VARIABLE + msgSeqCounter + +\* Partition assignment: 0 = no partition, 1..MaxPartitions = partition group +VARIABLE + partitions + +pendingMsgVars == <> + +messageVars == <> + +partitionVars == <> + +---- + +\* The server's term number. +VARIABLE + \* @type: Int -> Int; + currentTerm +\* The server's state (Follower, Candidate, or Leader). +VARIABLE + \* @type: Int -> Str; + state +\* The candidate the server voted for in its current term, or +\* Nil if it hasn't voted for any. +VARIABLE + \* @type: Int -> Int; + votedFor +serverVars == <> + +\* Server's log with offset-based indexing. +VARIABLE + \* @type: Int -> [ offset: Int, entries: LOGT, snapshotIndex: Int, snapshotTerm: Int ]; + log +\* Ghost variable: full log history for verification +VARIABLE + \* @type: Int -> LOGT; + historyLog +\* The highest log index the state machine may apply. +VARIABLE + \* @type: Int -> Int; + commitIndex +\* The highest log index applied to the state machine. +VARIABLE + \* @type: Int -> Int; + applied +logVars == <> + +\* Set of servers that responded to RequestVote in currentTerm. +VARIABLE + \* @type: Int -> Set(Int); + votesResponded +\* Set of servers that granted vote in currentTerm. +VARIABLE + \* @type: Int -> Set(Int); + votesGranted +\* @type: Seq(Int -> Set(Int)); +candidateVars == <> + +\* Highest log index each follower has acknowledged. +VARIABLE + \* @type: Int -> (Int -> Int); + matchIndex +VARIABLE + pendingConfChangeIndex +leaderVars == <> + +\* @type: Int -> [jointConfig: Seq(Set(int)), learners: Set(int)] +VARIABLE + config +VARIABLE + reconfigCount + +\* Index of the last applied config entry per server +VARIABLE + \* @type: Int -> Int; + appliedConfigIndex + +configVars == <> + +VARIABLE + durableState + +\* ============================================================================ +\* Progress state machine variables +\* Reference: tracker/progress.go:30-117 +\* ============================================================================ + +VARIABLE + \* @type: Int -> (Int -> Str); + progressState + +VARIABLE + \* @type: Int -> (Int -> Int); + pendingSnapshot + +VARIABLE + \* @type: Int -> (Int -> Int); + nextIndex + +VARIABLE + \* @type: Int -> (Int -> Bool); + msgAppFlowPaused + +\* ============================================================================ +\* Inflights flow control variables +\* Reference: tracker/inflights.go:28-40 +\* ============================================================================ + +VARIABLE + \* @type: Int -> (Int -> Set(Int)); + inflights \* Modeled as set (actual code uses ring buffer); sufficient for capacity checks + +progressVars == <> + +---- + +vars == <> + +\* View variables: excludes ghost/derived variables for state space reduction +view_vars == <> + +---- +\* Helpers + +\* The set of all quorums. This just calculates simple majorities, but the only +\* important property is that every quorum overlaps with every other. +Quorum(c) == {i \in SUBSET(c) : Cardinality(i) * 2 > Cardinality(c)} + +\* ============================================================================ +\* Network Partition Helpers +\* ============================================================================ + +\* Check if there is currently a network partition active +IsPartitioned == \E i \in Server : partitions[i] /= 0 + +\* Check if two servers can communicate (in the same partition or no partition) +CanCommunicate(i, j) == + \/ partitions[i] = 0 \* No partition active for i + \/ partitions[j] = 0 \* No partition active for j + \/ partitions[i] = partitions[j] \* Same partition + +\* Get all messages that cross partition boundaries (to be dropped) +CrossPartitionMessages == + {m \in DOMAIN messages : + /\ partitions[m.msource] /= 0 + /\ partitions[m.mdest] /= 0 + /\ partitions[m.msource] /= partitions[m.mdest]} + +\* Get all pending messages that cross partition boundaries +CrossPartitionPendingMessages == + {m \in DOMAIN pendingMessages : + /\ partitions[m.msource] /= 0 + /\ partitions[m.mdest] /= 0 + /\ partitions[m.msource] /= partitions[m.mdest]} + +\* Drop all messages between two servers +DropMessagesBetween(i, j) == + {m \in DOMAIN messages : (m.msource = i /\ m.mdest = j) \/ (m.msource = j /\ m.mdest = i)} + +\* Drop all messages to/from a specific server +MessagesToOrFrom(i) == + {m \in DOMAIN messages : m.msource = i \/ m.mdest = i} + +\* Drop all pending messages to/from a specific server +PendingMessagesToOrFrom(i) == + {m \in DOMAIN pendingMessages : m.msource = i \/ m.mdest = i} + +\* ============================================================================ +\* Virtual Log Helpers (Offset-aware) +\* ============================================================================ + +\* Get the logical index of the last entry in the log +\* @type: [ offset: Int, entries: LOGT, snapshotIndex: Int, snapshotTerm: Int ] => Int; +LastIndex(xlog) == + xlog.offset + Len(xlog.entries) - 1 + +\* Get the term of the last entry (or snapshot term if empty) +\* @type: [ offset: Int, entries: LOGT, snapshotIndex: Int, snapshotTerm: Int ] => Int; +LastTerm(xlog) == + IF Len(xlog.entries) > 0 THEN xlog.entries[Len(xlog.entries)].term + ELSE xlog.snapshotTerm + +\* Check if a logical index is available in the memory log (not compacted) +IsAvailable(i, index) == + /\ index >= log[i].offset + /\ index <= LastIndex(log[i]) + +\* Get the log entry at a logical index +\* PRECONDITION: IsAvailable(i, index) +LogEntry(i, index) == + log[i].entries[index - log[i].offset + 1] + +\* Get the term at a logical index; returns 0 for index 0 or unavailable indices. +LogTerm(i, index) == + IF index = 0 THEN 0 + ELSE IF index = log[i].snapshotIndex THEN log[i].snapshotTerm + ELSE IF IsAvailable(i, index) THEN LogEntry(i, index).term + ELSE 0 + +\* Reference: raft.go findConflictByTerm - largest index in [firstIndex-1, index] with term <= targetTerm +FindConflictByTerm(i, index, targetTerm) == + LET firstIdx == log[i].offset + validRange == (firstIdx - 1)..index + matchingIndices == {k \in validRange : LogTerm(i, k) <= targetTerm} + IN IF index = 0 \/ matchingIndices = {} THEN 0 ELSE Max(matchingIndices) + +\* Reference: raft.go handleAppendEntries - compute rejection hint via findConflictByTerm +ComputeRejectHint(i, rejectedIndex, leaderTerm) == + LET hintIndex == Min({rejectedIndex, LastIndex(log[i])}) + IN FindConflictByTerm(i, hintIndex, leaderTerm) + +\* Add message m to bag of messages. +\* @type: (MSG, MSG -> Int) => MSG -> Int; +WithMessage(m, msgs) == msgs (+) SetToBag({m}) + +\* Remove message m from bag of messages. +\* @type: (MSG, MSG -> Int) => MSG -> Int; +WithoutMessage(m, msgs) == msgs (-) SetToBag({m}) + +\* Add message to pendingMessages; adds sequence number when MsgNoReorder is TRUE. +SendDirect(m) == + IF MsgNoReorder + THEN /\ pendingMessages' = WithMessage(m @@ [mseq |-> msgSeqCounter], pendingMessages) + /\ msgSeqCounter' = msgSeqCounter + 1 + ELSE /\ pendingMessages' = WithMessage(m, pendingMessages) + /\ UNCHANGED msgSeqCounter + +\* All pending messages sent from node i +PendingMessages(i) == + FoldBag(LAMBDA x, y: IF y.msource = i THEN BagAdd(x,y) ELSE x, EmptyBag, pendingMessages) + +\* Remove all messages in pendingMessages that were sent from node i +ClearPendingMessages(i) == + pendingMessages (-) PendingMessages(i) + +\* Move all messages which was sent from node i in pendingMessages to messages +SendPendingMessages(i) == + LET msgs == PendingMessages(i) + IN /\ messages' = msgs (+) messages + /\ pendingMessages' = pendingMessages (-) msgs + +\* Remove a message from messages or pendingMessages. +DiscardDirect(m) == + IF m \in DOMAIN messages + THEN messages' = WithoutMessage(m, messages) /\ UNCHANGED <> + ELSE pendingMessages' = WithoutMessage(m, pendingMessages) /\ UNCHANGED <> + +\* Send response and discard request; adds sequence number when MsgNoReorder is TRUE. +ReplyDirect(response, request) == + LET resp == IF MsgNoReorder THEN response @@ [mseq |-> msgSeqCounter] ELSE response + IN IF request \in DOMAIN messages + THEN /\ messages' = WithoutMessage(request, messages) + /\ pendingMessages' = WithMessage(resp, pendingMessages) + /\ IF MsgNoReorder THEN msgSeqCounter' = msgSeqCounter + 1 ELSE UNCHANGED msgSeqCounter + ELSE /\ pendingMessages' = WithMessage(resp, WithoutMessage(request, pendingMessages)) + /\ UNCHANGED messages + /\ IF MsgNoReorder THEN msgSeqCounter' = msgSeqCounter + 1 ELSE UNCHANGED msgSeqCounter + +\* Default: change when needed + Send(m) == SendDirect(m) + Reply(response, request) == ReplyDirect(response, request) + Discard(m) == DiscardDirect(m) + +\* Check if m is the lowest-sequence message from source to dest (FIFO ordering). +IsFifoFirst(m) == + IF ~MsgNoReorder THEN TRUE + ELSE ~\E other \in DOMAIN messages: + /\ other.msource = m.msource + /\ other.mdest = m.mdest + /\ other.mseq < m.mseq + +MaxOrZero(s) == IF s = {} THEN 0 ELSE Max(s) + +GetJointConfig(i) == + config[i].jointConfig + +GetConfig(i) == + GetJointConfig(i)[1] + +GetOutgoingConfig(i) == + GetJointConfig(i)[2] + +IsJointConfig(i) == + /\ GetJointConfig(i)[2] # {} + +GetLearners(i) == + config[i].learners + +\* Compute ConfState from history sequence for snapshot metadata. +\* @type: Seq([value: a, term: Int, type: Str]) => [voters: Set(Str), learners: Set(Str), outgoing: Set(Str), autoLeave: Bool]; +ComputeConfStateFromHistory(history) == + LET configIndices == {k \in 1..Len(history) : history[k].type = ConfigEntry} + lastConfigIdx == IF configIndices /= {} THEN Max(configIndices) ELSE 0 + IN + IF lastConfigIdx = 0 THEN + [voters |-> {}, learners |-> {}, outgoing |-> {}, autoLeave |-> FALSE] + ELSE + LET entry == history[lastConfigIdx] + isLeaveJoint == "leaveJoint" \in DOMAIN entry.value /\ entry.value.leaveJoint = TRUE + IN + IF isLeaveJoint THEN + [voters |-> entry.value.newconf, + learners |-> IF "learners" \in DOMAIN entry.value THEN entry.value.learners ELSE {}, + outgoing |-> {}, + autoLeave |-> FALSE] + ELSE + LET hasEnterJoint == "enterJoint" \in DOMAIN entry.value + enterJoint == IF hasEnterJoint THEN entry.value.enterJoint ELSE FALSE + hasOldconf == enterJoint /\ "oldconf" \in DOMAIN entry.value + oldconf == IF hasOldconf THEN entry.value.oldconf ELSE {} + IN + [voters |-> entry.value.newconf, + learners |-> IF "learners" \in DOMAIN entry.value THEN entry.value.learners ELSE {}, + outgoing |-> oldconf, + autoLeave |-> enterJoint /\ oldconf /= {}] + +\* Reference: raft.go applyConfChange(), confchange.go LeaveJoint() +ApplyConfigUpdate(i, k) == + LET entry == LogEntry(i, k) + isLeaveJoint == "leaveJoint" \in DOMAIN entry.value /\ entry.value.leaveJoint = TRUE + newVoters == IF isLeaveJoint THEN GetConfig(i) ELSE entry.value.newconf + newLearners == IF "learners" \in DOMAIN entry.value THEN entry.value.learners ELSE {} + enterJoint == IF "enterJoint" \in DOMAIN entry.value THEN entry.value.enterJoint ELSE FALSE + outgoing == IF enterJoint THEN entry.value.oldconf ELSE {} + newAutoLeave == IF isLeaveJoint THEN FALSE ELSE enterJoint + IN + [config EXCEPT ![i]= [jointConfig |-> << newVoters, outgoing >>, learners |-> newLearners, autoLeave |-> newAutoLeave]] + +\* Apply a single config change to a config record. +\* @type: ([nid: Str, action: Str], [voters: Set(Str), learners: Set(Str)]) => [voters: Set(Str), learners: Set(Str)]; +ApplyChange(change, conf) == + CASE change.action = "AddNewServer" -> + [voters |-> conf.voters \union {change.nid}, + learners |-> conf.learners \ {change.nid}] + [] change.action = "RemoveServer" -> + [voters |-> conf.voters \ {change.nid}, + learners |-> conf.learners \ {change.nid}] + [] change.action = "AddLearner" -> + [voters |-> conf.voters \ {change.nid}, + learners |-> conf.learners \union {change.nid}] + [] OTHER -> conf + +CommitTo(i, c) == + commitIndex' = [commitIndex EXCEPT ![i] = Max({@, c})] + +CurrentLeaders == {i \in Server : state[i] = Leader} + +PersistState(i) == + durableState' = [durableState EXCEPT ![i] = [ + currentTerm |-> currentTerm[i], + votedFor |-> votedFor[i], + log |-> LastIndex(log[i]), + entries |-> log[i].entries, + snapshotIndex |-> log[i].snapshotIndex, + snapshotTerm |-> log[i].snapshotTerm, + snapshotHistory |-> @.snapshotHistory, + commitIndex |-> commitIndex[i], + config |-> config[i] + ]] + +\* ============================================================================ +\* Progress and Inflights helper functions +\* ============================================================================ + +\* Reference: inflights.go Count() +InflightsCount(i, j) == Cardinality(inflights[i][j]) + +\* Reference: inflights.go Full() +InflightsFull(i, j) == InflightsCount(i, j) >= MaxInflightMsgs + +\* Reference: progress.go IsPaused() - checks msgAppFlowPaused flag +IsPaused(i, j) == + CASE progressState[i][j] = StateProbe -> msgAppFlowPaused[i][j] + [] progressState[i][j] = StateReplicate -> msgAppFlowPaused[i][j] + [] progressState[i][j] = StateSnapshot -> TRUE + [] OTHER -> FALSE + +\* Reference: inflights.go Add() +AddInflight(i, j, lastIndex) == + inflights' = [inflights EXCEPT ![i][j] = @ \cup {lastIndex}] + +\* Reference: inflights.go FreeLE() +FreeInflightsLE(i, j, index) == + inflights' = [inflights EXCEPT ![i][j] = {idx \in @ : idx > index}] + +\* Reference: progress.go ResetState() +ResetInflights(i, j) == + inflights' = [inflights EXCEPT ![i][j] = {}] + +---- +InitMessageVars == /\ messages = EmptyBag + /\ pendingMessages = EmptyBag + /\ msgSeqCounter = 0 +InitServerVars == /\ currentTerm = [i \in Server |-> 0] + /\ state = [i \in Server |-> Follower] + /\ votedFor = [i \in Server |-> Nil] +InitCandidateVars == /\ votesResponded = [i \in Server |-> {}] + /\ votesGranted = [i \in Server |-> {}] +InitLeaderVars == /\ matchIndex = [i \in Server |-> [j \in Server |-> 0]] + /\ pendingConfChangeIndex = [i \in Server |-> 0] +InitLogVars == /\ log = [i \in Server |-> [offset |-> 1, entries |-> <<>>, snapshotIndex |-> 0, snapshotTerm |-> 0]] + /\ historyLog = [i \in Server |-> <<>>] + /\ commitIndex = [i \in Server |-> 0] + /\ applied = [i \in Server |-> 0] +InitConfigVars == /\ config = [i \in Server |-> [ jointConfig |-> <>, learners |-> {}, autoLeave |-> FALSE]] + /\ reconfigCount = 0 + /\ appliedConfigIndex = [i \in Server |-> 0] +InitDurableState == + durableState = [ i \in Server |-> [ + currentTerm |-> currentTerm[i], + votedFor |-> votedFor[i], + log |-> 0, + entries |-> <<>>, + snapshotIndex |-> 0, + snapshotTerm |-> 0, + snapshotHistory |-> <<>>, + commitIndex |-> commitIndex[i], + config |-> config[i] + ]] + +\* Reference: raft.go becomeFollower - all Progress starts at StateProbe +InitProgressVars == + /\ progressState = [i \in Server |-> [j \in Server |-> StateProbe]] + /\ pendingSnapshot = [i \in Server |-> [j \in Server |-> 0]] + /\ nextIndex = [i \in Server |-> [j \in Server |-> 1]] + /\ msgAppFlowPaused = [i \in Server |-> [j \in Server |-> FALSE]] + /\ inflights = [i \in Server |-> [j \in Server |-> {}]] + +\* No partition active initially. +InitPartitionVars == + /\ partitions = [i \in Server |-> 0] + +Init == /\ InitMessageVars + /\ InitServerVars + /\ InitCandidateVars + /\ InitLeaderVars + /\ InitLogVars + /\ InitConfigVars + /\ InitDurableState + /\ InitProgressVars + /\ InitPartitionVars + +---- +\* Define state transitions + +\* Server i restarts from stable storage, losing all volatile state. +\* @type: Int => Bool; +Restart(i) == + /\ state' = [state EXCEPT ![i] = Follower] + /\ votesResponded' = [votesResponded EXCEPT ![i] = {}] + /\ votesGranted' = [votesGranted EXCEPT ![i] = {}] + /\ matchIndex' = [matchIndex EXCEPT ![i] = [j \in Server |-> 0]] + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i] = 0] + /\ pendingMessages' = ClearPendingMessages(i) + /\ currentTerm' = [currentTerm EXCEPT ![i] = durableState[i].currentTerm] + /\ commitIndex' = [commitIndex EXCEPT ![i] = durableState[i].commitIndex] + /\ votedFor' = [votedFor EXCEPT ![i] = durableState[i].votedFor] + /\ log' = [log EXCEPT ![i] = [ + offset |-> durableState[i].snapshotIndex + 1, + entries |-> durableState[i].entries, + snapshotIndex |-> durableState[i].snapshotIndex, + snapshotTerm |-> durableState[i].snapshotTerm + ]] + /\ applied' = [applied EXCEPT ![i] = durableState[i].snapshotIndex] + /\ config' = [config EXCEPT ![i] = durableState[i].config] + /\ appliedConfigIndex' = [appliedConfigIndex EXCEPT ![i] = durableState[i].commitIndex] + /\ progressState' = [progressState EXCEPT ![i] = [j \in Server |-> StateProbe]] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i] = [j \in Server |-> FALSE]] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i] = [j \in Server |-> 0]] + /\ nextIndex' = [nextIndex EXCEPT ![i] = [j \in Server |-> 1]] + /\ inflights' = [inflights EXCEPT ![i] = [j \in Server |-> {}]] + /\ IF RestartDropsMessages + THEN messages' = messages (-) SetToBag(MessagesToOrFrom(i)) + ELSE UNCHANGED messages + \* historyLog restored from durable state (entries beyond snapshot lost on crash) + /\ historyLog' = [historyLog EXCEPT ![i] = durableState[i].snapshotHistory \o durableState[i].entries] + /\ UNCHANGED <> + +\* Server i times out and starts a new election. +\* @type: Int => Bool; +Timeout(i) == /\ state[i] \in {Follower, Candidate} + /\ i \in GetConfig(i) \union GetOutgoingConfig(i) + /\ state' = [state EXCEPT ![i] = Candidate] + /\ currentTerm' = [currentTerm EXCEPT ![i] = currentTerm[i] + 1] + /\ votedFor' = [votedFor EXCEPT ![i] = i] + /\ votesResponded' = [votesResponded EXCEPT ![i] = {}] + /\ votesGranted' = [votesGranted EXCEPT ![i] = {}] + /\ UNCHANGED <> + +\* Candidate i sends j a RequestVote request. +\* @type: (Int, Int) => Bool; +RequestVote(i, j) == + /\ state[i] = Candidate + /\ j \in ((GetConfig(i) \union GetOutgoingConfig(i) \union GetLearners(i)) \ votesResponded[i]) + /\ IF i # j + THEN Send([mtype |-> RequestVoteRequest, + mterm |-> currentTerm[i], + mlastLogTerm |-> LastTerm(log[i]), + mlastLogIndex |-> LastIndex(log[i]), + msource |-> i, + mdest |-> j]) + ELSE Send([mtype |-> RequestVoteResponse, + mterm |-> currentTerm[i], + mvoteGranted |-> TRUE, + msource |-> i, + mdest |-> i]) + /\ UNCHANGED <> + +\* Leader i sends j an AppendEntries request containing entries in [b,e) range. +\* N.B. range is right open +\* @type: (Int, Int, <>, Int) => Bool; +AppendEntriesInRangeToPeer(subtype, i, j, range) == + /\ i /= j + /\ range[1] <= range[2] + /\ state[i] = Leader + /\ j \in GetConfig(i) \union GetOutgoingConfig(i) \union GetLearners(i) + \* Entries must be available (not compacted); heartbeat skips this check + /\ (range[1] = range[2] \/ range[1] >= log[i].offset) + \* prevLogIndex must have retrievable term (Bug 76f1249 fix) + /\ (range[1] = 1 \/ range[1] - 1 = log[i].snapshotIndex \/ range[1] - 1 >= log[i].offset) + \* Heartbeat bypasses flow control (Reference: raft.go bcastHeartbeat()) + /\ (subtype = "heartbeat" \/ ~IsPaused(i, j)) + /\ LET + prevLogIndex == range[1] - 1 + \* Upper bound simplifies verification. + prevLogTerm == IF prevLogIndex > 0 /\ prevLogIndex <= LastIndex(log[i]) THEN + LogTerm(i, prevLogIndex) + ELSE + 0 + lastEntry == Min({LastIndex(log[i]), range[2]-1}) + entries == SubSeq(log[i].entries, range[1] - log[i].offset + 1, lastEntry - log[i].offset + 1) + \* Commit bounded differently per subtype (Reference: raft.go sendAppend()) + commit == CASE subtype = "heartbeat" -> Min({commitIndex[i], matchIndex[i][j]}) + [] lastEntry < range[1] -> commitIndex[i] + [] OTHER -> Min({commitIndex[i], lastEntry}) + numEntries == Len(entries) + newInflights == IF lastEntry >= range[1] + THEN inflights[i][j] \cup {lastEntry} + ELSE inflights[i][j] + \* Reference: progress.go SentEntries() / SentCommit() + newMsgAppFlowPaused == + CASE subtype = "heartbeat" + -> msgAppFlowPaused[i][j] + [] progressState[i][j] = StateReplicate + -> Cardinality(newInflights) >= MaxInflightMsgs + [] progressState[i][j] = StateProbe /\ numEntries > 0 + -> TRUE + [] OTHER -> msgAppFlowPaused[i][j] + IN /\ Send( [mtype |-> AppendEntriesRequest, + msubtype |-> subtype, + mterm |-> currentTerm[i], + mprevLogIndex |-> prevLogIndex, + mprevLogTerm |-> prevLogTerm, + mentries |-> entries, + mcommitIndex |-> commit, + msource |-> i, + mdest |-> j]) + \* Inflights only added in StateReplicate for non-heartbeat (Reference: progress.go SentEntries()) + /\ IF lastEntry >= range[1] /\ subtype /= "heartbeat" /\ progressState[i][j] = StateReplicate + THEN AddInflight(i, j, lastEntry) + ELSE UNCHANGED inflights + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = newMsgAppFlowPaused] + \* nextIndex advanced only in StateReplicate + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = + IF numEntries > 0 /\ progressState[i][j] = StateReplicate /\ subtype /= "heartbeat" + THEN @ + numEntries + ELSE @] + /\ UNCHANGED <> + /\ UNCHANGED <> + +\* etcd leader sends MsgAppResp to itself immediately after appending log entry +AppendEntriesToSelf(i) == + /\ state[i] = Leader + /\ Send([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> LastIndex(log[i]), + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> i]) + /\ UNCHANGED <> + +AppendEntries(i, j, range) == + AppendEntriesInRangeToPeer("app", i, j, range) + +Heartbeat(i, j) == + \* heartbeat is equivalent to an append-entry request with 0 entry index 1 + AppendEntriesInRangeToPeer("heartbeat", i, j, <<1,1>>) + +SendSnapshot(i, j) == + /\ i /= j + /\ state[i] = Leader + /\ j \in GetConfig(i) \union GetLearners(i) + \* prevLogIndex not available, must send snapshot instead + /\ LET prevLogIndex == nextIndex[i][j] - 1 IN + ~IsAvailable(i, prevLogIndex) + /\ log[i].snapshotIndex > 0 + /\ LET snapshotHistory == durableState[i].snapshotHistory + IN Send([mtype |-> SnapshotRequest, + mterm |-> currentTerm[i], + msnapshotIndex |-> log[i].snapshotIndex, + msnapshotTerm |-> log[i].snapshotTerm, + mhistory |-> snapshotHistory, + mconfState |-> ComputeConfStateFromHistory(snapshotHistory), + msource |-> i, + mdest |-> j]) + \* Reference: raft.go sendSnapshot() -> pr.BecomeSnapshot() + /\ progressState' = [progressState EXCEPT ![i][j] = StateSnapshot] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i][j] = log[i].snapshotIndex] + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = log[i].snapshotIndex + 1] + /\ ResetInflights(i, j) + /\ UNCHANGED <> + +\* Send snapshot with compaction. Reference: raft.go maybeSendSnapshot() +SendSnapshotWithCompaction(i, j, snapshoti) == + /\ i /= j + /\ state[i] = Leader + /\ j \in GetConfig(i) \union GetLearners(i) + /\ snapshoti <= applied[i] + /\ snapshoti >= log[i].snapshotIndex + /\ snapshoti >= matchIndex[i][j] + /\ LET snapshotHistory == IF snapshoti <= durableState[i].snapshotIndex + THEN SubSeq(durableState[i].snapshotHistory, 1, snapshoti) + ELSE durableState[i].snapshotHistory \o + SubSeq(log[i].entries, 1, snapshoti - log[i].offset + 1) + IN SendDirect([mtype |-> SnapshotRequest, + mterm |-> currentTerm[i], + msnapshotIndex |-> snapshoti, + msnapshotTerm |-> LogTerm(i, snapshoti), + mhistory |-> snapshotHistory, + mconfState |-> ComputeConfStateFromHistory(snapshotHistory), + msource |-> i, + mdest |-> j]) + /\ progressState' = [progressState EXCEPT ![i][j] = StateSnapshot] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i][j] = snapshoti] + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = snapshoti + 1] + /\ inflights' = [inflights EXCEPT ![i][j] = {}] + /\ UNCHANGED <> + +\* Test harness snapshot injection: bypasses normal raft path, does NOT modify progress state. +\* Creates snapshot from applied state. Reference: rafttest/interaction_env_handler_send_snapshot.go +ManualSendSnapshot(i, j) == + /\ i /= j + /\ state[i] = Leader + /\ j \in GetConfig(i) \union GetLearners(i) + /\ applied[i] > 0 + /\ LET snapshotHistory == IF applied[i] <= durableState[i].snapshotIndex + THEN SubSeq(durableState[i].snapshotHistory, 1, applied[i]) + ELSE durableState[i].snapshotHistory \o + SubSeq(log[i].entries, 1, applied[i] - log[i].offset + 1) + IN Send([mtype |-> SnapshotRequest, + mterm |-> currentTerm[i], + msnapshotIndex |-> applied[i], + msnapshotTerm |-> LogTerm(i, applied[i]), + mhistory |-> snapshotHistory, + mconfState |-> ComputeConfStateFromHistory(snapshotHistory), + msource |-> i, + mdest |-> j]) + /\ UNCHANGED <> + +\* Candidate i transitions to leader. +\* @type: Int => Bool; +BecomeLeader(i) == + /\ state[i] = Candidate + /\ IF IsJointConfig(i) THEN + /\ (votesGranted[i] \cap GetConfig(i)) \in Quorum(GetConfig(i)) + /\ (votesGranted[i] \cap GetOutgoingConfig(i)) \in Quorum(GetOutgoingConfig(i)) + ELSE + votesGranted[i] \in Quorum(GetConfig(i)) + /\ state' = [state EXCEPT ![i] = Leader] + /\ matchIndex' = [matchIndex EXCEPT ![i] = + [j \in Server |-> IF j = i THEN LastIndex(log[i]) ELSE 0]] + \* Reference: raft.go becomeLeader() -> reset() + /\ progressState' = [progressState EXCEPT ![i] = + [j \in Server |-> IF j = i THEN StateReplicate ELSE StateProbe]] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i] = + [j \in Server |-> FALSE]] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i] = + [j \in Server |-> 0]] + /\ nextIndex' = [nextIndex EXCEPT ![i] = + [j \in Server |-> LastIndex(log[i]) + 1]] + /\ inflights' = [inflights EXCEPT ![i] = + [j \in Server |-> {}]] + \* Conservatively delay future proposals until pending entries committed (raft.go:955-960) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i] = LastIndex(log[i])] + /\ UNCHANGED <> + +Replicate(i, v, t) == + /\ t \in {ValueEntry, ConfigEntry} + /\ state[i] = Leader + /\ LET entry == [term |-> currentTerm[i], + type |-> t, + value |-> v] + newLog == [log[i] EXCEPT !.entries = Append(@, entry)] + IN /\ log' = [log EXCEPT ![i] = newLog] + /\ historyLog' = [historyLog EXCEPT ![i] = Append(@, entry)] + +\* Leader i receives a client request to add v to the log. +\* @type: (Int, Int) => Bool; +ClientRequest(i, v) == + /\ Replicate(i, [val |-> v], ValueEntry) + /\ UNCHANGED <> + +\* ClientRequest variant that also sends MsgAppResp (for trace validation). +ClientRequestAndSend(i, v) == + /\ Replicate(i, [val |-> v], ValueEntry) + /\ Send([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> LastIndex(log'[i]), + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> i]) + /\ UNCHANGED <> + + +\* Leader replicates an implicit entry (self-message response in trace). +\* In joint config with auto-leave: creates LeaveJoint entry. Otherwise: normal value entry. +\* Also supports DisableConfChangeValidation (Reference: raft.go:1762-1770) +ReplicateImplicitEntry(i) == + /\ state[i] = Leader + /\ LET isJoint == IsJointConfig(i) + pendingIdx == pendingConfChangeIndex[i] + hasPendingEnterJoint == + /\ pendingIdx > 0 + /\ pendingIdx > log[i].offset + /\ pendingIdx <= LastIndex(log[i]) + /\ LET entry == LogEntry(i, pendingIdx) + IN /\ entry.type = ConfigEntry + /\ "enterJoint" \in DOMAIN entry.value + /\ entry.value.enterJoint = TRUE + autoLeaveCondition == + /\ isJoint + /\ config[i].autoLeave = TRUE + /\ applied[i] >= pendingConfChangeIndex[i] + \* Reference: raft.go:1334-1338 - bypasses "not in joint state" check + disableValidationCondition == + /\ DisableConfChangeValidation + /\ ~isJoint + /\ hasPendingEnterJoint + shouldCreateLeaveJoint == autoLeaveCondition \/ disableValidationCondition + \* disableValidation: learners from pending entry (not yet applied) + leaveJointLearners == IF disableValidationCondition + THEN LogEntry(i, pendingIdx).value.learners + ELSE GetLearners(i) + IN + /\ (isJoint => (config[i].autoLeave = TRUE /\ applied[i] >= pendingConfChangeIndex[i])) + /\ LET entryType == IF shouldCreateLeaveJoint THEN ConfigEntry ELSE ValueEntry + entryValue == IF shouldCreateLeaveJoint + THEN [leaveJoint |-> TRUE, newconf |-> GetConfig(i), learners |-> leaveJointLearners] + ELSE [val |-> 0] + IN + /\ Replicate(i, entryValue, entryType) + /\ Send([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> LastIndex(log'[i]), + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> i]) + /\ IF shouldCreateLeaveJoint + THEN pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i] = LastIndex(log'[i])] + ELSE UNCHANGED pendingConfChangeIndex + /\ UNCHANGED <> + +\* Leader i advances its commitIndex using current applied config's quorum. +\* Reference: raft.go maybeCommit() -> trk.Committed() +\* @type: Int => Bool; +AdvanceCommitIndex(i) == + /\ state[i] = Leader + /\ LET AllVoters == GetConfig(i) \union GetOutgoingConfig(i) + Agree(index) == {k \in AllVoters : matchIndex[i][k] >= index} + logSize == LastIndex(log[i]) + IsCommitted(index) == + IF IsJointConfig(i) THEN + /\ (Agree(index) \cap GetConfig(i)) \in Quorum(GetConfig(i)) + /\ (Agree(index) \cap GetOutgoingConfig(i)) \in Quorum(GetOutgoingConfig(i)) + ELSE + Agree(index) \in Quorum(GetConfig(i)) + + agreeIndexes == {index \in (commitIndex[i]+1)..logSize : IsCommitted(index)} + newCommitIndex == + IF /\ agreeIndexes /= {} + /\ LogTerm(i, Max(agreeIndexes)) = currentTerm[i] + THEN + Max(agreeIndexes) + ELSE + commitIndex[i] + IN + /\ CommitTo(i, newCommitIndex) + /\ UNCHANGED <> + + +\* Leader i adds a new server j or promote learner j +AddNewServer(i, j) == + /\ state[i] = Leader + /\ j \notin GetConfig(i) + /\ ~IsJointConfig(i) + /\ IF pendingConfChangeIndex[i] = 0 THEN + /\ Replicate(i, [newconf |-> GetConfig(i) \union {j}, learners |-> GetLearners(i)], ConfigEntry) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i]=LastIndex(log'[i])] + ELSE + /\ Replicate(i, <<>>, ValueEntry) + /\ UNCHANGED <> + /\ UNCHANGED <> + +\* Leader i adds a leaner j to the cluster. +AddLearner(i, j) == + /\ state[i] = Leader + /\ j \notin GetConfig(i) \union GetLearners(i) + /\ ~IsJointConfig(i) + /\ IF pendingConfChangeIndex[i] = 0 THEN + /\ Replicate(i, [newconf |-> GetConfig(i), learners |-> GetLearners(i) \union {j}], ConfigEntry) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i]=LastIndex(log'[i])] + ELSE + /\ Replicate(i, <<>>, ValueEntry) + /\ UNCHANGED <> + /\ UNCHANGED <> + +\* Leader i removes a server j (possibly itself) from the cluster. +DeleteServer(i, j) == + /\ state[i] = Leader + /\ j \in GetConfig(i) \union GetLearners(i) + /\ ~IsJointConfig(i) + /\ IF pendingConfChangeIndex[i] = 0 THEN + /\ Replicate(i, [newconf |-> GetConfig(i) \ {j}, learners |-> GetLearners(i) \ {j}], ConfigEntry) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i]=LastIndex(log'[i])] + ELSE + /\ Replicate(i, <<>>, ValueEntry) + /\ UNCHANGED <> + /\ UNCHANGED <> + +\* Leader i proposes a configuration change. Reference: confchange/confchange.go +ChangeConf(i) == + /\ state[i] = Leader + /\ IF pendingConfChangeIndex[i] <= applied[i] THEN + \E newVoters \in SUBSET Server, newLearners \in SUBSET Server, enterJoint \in {TRUE, FALSE}: + /\ ~IsJointConfig(i) + \* Simple change: at most one voter changed + /\ (enterJoint = FALSE) => + Cardinality((GetConfig(i) \ newVoters) \union (newVoters \ GetConfig(i))) <= 1 + /\ newVoters \cap newLearners = {} + /\ (enterJoint = TRUE) => (GetConfig(i) \cap newLearners = {}) + /\ newVoters /= {} + /\ Replicate(i, [newconf |-> newVoters, learners |-> newLearners, enterJoint |-> enterJoint, oldconf |-> GetConfig(i)], ConfigEntry) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i]=LastIndex(log'[i])] + ELSE + /\ Replicate(i, <<>>, ValueEntry) + /\ UNCHANGED <> + /\ UNCHANGED <> + +\* ChangeConf variant that also sends MsgAppResp (for trace validation). +ChangeConfAndSend(i) == + /\ state[i] = Leader + /\ IF pendingConfChangeIndex[i] = 0 THEN + \E newVoters \in SUBSET Server, newLearners \in SUBSET Server, enterJoint \in {TRUE, FALSE}: + /\ ~IsJointConfig(i) + /\ (enterJoint = FALSE) => + Cardinality((GetConfig(i) \ newVoters) \union (newVoters \ GetConfig(i))) <= 1 + /\ newVoters \cap newLearners = {} + /\ (enterJoint = TRUE) => (GetConfig(i) \cap newLearners = {}) + /\ newVoters /= {} + /\ Replicate(i, [newconf |-> newVoters, learners |-> newLearners, enterJoint |-> enterJoint, oldconf |-> GetConfig(i)], ConfigEntry) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i]=LastIndex(log'[i])] + /\ Send([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> LastIndex(log'[i]), + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> i]) + ELSE + /\ Replicate(i, <<>>, ValueEntry) + /\ UNCHANGED <> + /\ Send([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> LastIndex(log'[i]), + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> i]) + /\ UNCHANGED <> + +\* Apply the next committed config entry in order +\* Reference: etcd processes CommittedEntries sequentially (for _, entry := range rd.CommittedEntries) +\* and calls ApplyConfChange for each config entry as it's encountered +\* This ensures configs are applied one at a time, in log order +ApplySimpleConfChange(i) == + \* Find config entries that are committed but not yet applied + LET validIndices == {x \in Max({log[i].offset, appliedConfigIndex[i]+1})..commitIndex[i] : + LogEntry(i, x).type = ConfigEntry} + IN + /\ validIndices /= {} + /\ LET k == Min(validIndices) \* Apply the NEXT config entry, not MAX + oldConfig == GetConfig(i) \cup GetOutgoingConfig(i) \cup GetLearners(i) + newConfigFn == ApplyConfigUpdate(i, k) + newConfig == newConfigFn[i].jointConfig[1] \cup newConfigFn[i].jointConfig[2] \cup newConfigFn[i].learners + addedNodes == newConfig \ oldConfig + IN + /\ k > 0 + /\ k <= commitIndex[i] + /\ config' = newConfigFn + /\ appliedConfigIndex' = [appliedConfigIndex EXCEPT ![i] = k] \* Track applied config + /\ IF state[i] = Leader /\ pendingConfChangeIndex[i] = k THEN + /\ reconfigCount' = reconfigCount + 1 + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i] = 0] + ELSE UNCHANGED <> + /\ IF state[i] = Leader /\ addedNodes # {} + THEN /\ nextIndex' = [nextIndex EXCEPT ![i] = + [j \in Server |-> IF j \in addedNodes THEN Max({LastIndex(log[i]), 1}) ELSE nextIndex[i][j]]] + \* Reference: confchange/confchange.go makeVoter() - only init Match=0 for truly new nodes + \* Existing nodes (including leader itself) keep their Match value (pr != nil check) + /\ matchIndex' = [matchIndex EXCEPT ![i] = + [j \in Server |-> IF j \in addedNodes /\ j # i THEN 0 ELSE matchIndex[i][j]]] + /\ progressState' = [progressState EXCEPT ![i] = + [j \in Server |-> IF j \in addedNodes /\ j # i THEN StateProbe ELSE progressState[i][j]]] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i] = + [j \in Server |-> IF j \in addedNodes THEN FALSE ELSE msgAppFlowPaused[i][j]]] + /\ inflights' = [inflights EXCEPT ![i] = + [j \in Server |-> IF j \in addedNodes THEN {} ELSE inflights[i][j]]] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i] = + [j \in Server |-> IF j \in addedNodes THEN 0 ELSE pendingSnapshot[i][j]]] + ELSE /\ UNCHANGED progressVars + /\ UNCHANGED matchIndex + /\ UNCHANGED <> + +\* Leave joint consensus - transition from joint config to single config +\* This action is called when applying a LeaveJoint config entry (via ApplySimpleConfChange) +\* Reference: confchange/confchange.go:94-121 LeaveJoint() +\* LeaveJoint preserves Learners and moves LearnersNext into Learners +LeaveJoint(i) == + /\ IsJointConfig(i) + /\ LET newVoters == GetConfig(i) \* Keep incoming config (jointConfig[1]) + \* Preserve learners (Reference: confchange.go:103-107 - LearnersNext are added to Learners) + IN config' = [config EXCEPT ![i] = [learners |-> GetLearners(i), jointConfig |-> <>, autoLeave |-> FALSE]] + /\ IF state[i] = Leader /\ pendingConfChangeIndex[i] > 0 THEN + /\ reconfigCount' = reconfigCount + 1 + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i] = 0] + ELSE UNCHANGED <> + /\ UNCHANGED <> + +\* Leader proposes LeaveJoint entry when autoLeave=TRUE and config entry is applied +\* Reference: raft.go:745-760 - AutoLeave mechanism +\* When r.trk.Config.AutoLeave && newApplied >= r.pendingConfIndex && r.state == StateLeader, +\* the leader automatically proposes an empty ConfChangeV2 to leave joint config. +\* This empty ConfChangeV2 must be committed with joint config's two quorums before applying. +ProposeLeaveJoint(i) == + /\ state[i] = Leader + /\ IsJointConfig(i) + /\ config[i].autoLeave = TRUE + /\ pendingConfChangeIndex[i] = 0 \* Previous config change has been applied + \* Propose a LeaveJoint config entry - represented as ConfigEntry with leaveJoint=TRUE + \* This entry must be committed with joint quorum before being applied + \* Reference: confchange.go:103-107 - LeaveJoint preserves Learners and adds LearnersNext + /\ Replicate(i, [leaveJoint |-> TRUE, newconf |-> GetConfig(i), learners |-> GetLearners(i)], ConfigEntry) + /\ pendingConfChangeIndex' = [pendingConfChangeIndex EXCEPT ![i] = LastIndex(log'[i])] + /\ UNCHANGED <> + +\* Apply configuration from snapshot +\* When a follower receives a snapshot, it applies the config directly +\* Use persisted snapshotHistory instead of ghost variable historyLog +\* Reference: confchange/restore.go - Restore() uses ConfState which contains Voters, Learners, VotersOutgoing, AutoLeave +ApplySnapshotConfChange(i, newVoters) == + \* Find the last config entry in persisted snapshotHistory to determine joint config + LET snapshotHist == durableState[i].snapshotHistory + configIndices == {k \in 1..Len(snapshotHist) : snapshotHist[k].type = ConfigEntry} + lastConfigIdx == IF configIndices /= {} THEN Max(configIndices) ELSE 0 + entry == IF lastConfigIdx > 0 THEN snapshotHist[lastConfigIdx] ELSE [value |-> [newconf |-> {}, learners |-> {}]] + \* Check if this is a leaveJoint entry + isLeaveJoint == "leaveJoint" \in DOMAIN entry.value /\ entry.value.leaveJoint = TRUE + \* Check if last config entry has enterJoint=TRUE + hasEnterJoint == ~isLeaveJoint /\ "enterJoint" \in DOMAIN entry.value + enterJoint == IF hasEnterJoint THEN entry.value.enterJoint ELSE FALSE + hasOldconf == enterJoint /\ "oldconf" \in DOMAIN entry.value + oldconf == IF hasOldconf THEN entry.value.oldconf ELSE {} + \* Read learners from entry (Reference: confchange/restore.go:82-87) + hasLearners == lastConfigIdx > 0 /\ "learners" \in DOMAIN entry.value + newLearners == IF hasLearners THEN entry.value.learners ELSE {} + \* AutoLeave is TRUE when entering joint config (snapshot may contain joint config) + \* For leaveJoint, autoLeave should be FALSE + newAutoLeave == ~isLeaveJoint /\ enterJoint /\ oldconf /= {} + IN + /\ config' = [config EXCEPT ![i] = [learners |-> newLearners, jointConfig |-> <>, autoLeave |-> newAutoLeave]] + /\ appliedConfigIndex' = [appliedConfigIndex EXCEPT ![i] = lastConfigIdx] + /\ UNCHANGED <> + +\* Apply committed entries to state machine +\* Reference: raft/node.go - application layer retrieves CommittedEntries from Ready() +\* and calls Advance() after applying them +\* This advances 'applied' from its current value up to any point <= commitIndex +\* Invariant: applied <= commitIndex (AppliedBoundInv) +ApplyEntries(i, newApplied) == + /\ newApplied > applied[i] \* Must make progress + /\ newApplied <= commitIndex[i] \* Cannot apply beyond committed + /\ applied' = [applied EXCEPT ![i] = newApplied] + /\ UNCHANGED <> + +Ready(i) == + /\ PersistState(i) + /\ SendPendingMessages(i) + /\ UNCHANGED <> + +BecomeFollowerOfTerm(i, t) == + /\ currentTerm' = [currentTerm EXCEPT ![i] = t] + /\ state' = [state EXCEPT ![i] = Follower] + /\ IF currentTerm[i] # t THEN + votedFor' = [votedFor EXCEPT ![i] = Nil] + ELSE + UNCHANGED <> + +StepDownToFollower(i) == + /\ state[i] \in {Leader, Candidate} + /\ BecomeFollowerOfTerm(i, currentTerm[i]) + /\ UNCHANGED <> + +\* Clear MsgAppFlowPaused on successful response. Reference: progress.go MaybeUpdate() +ClearMsgAppFlowPausedOnUpdate(i, j) == + msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + +---- +\* Message handlers +\* i = recipient, j = sender, m = message + +\* Server i receives a RequestVote request from server j with +\* m.mterm <= currentTerm[i]. +\* @type: (Int, Int, RVREQT) => Bool; +HandleRequestVoteRequest(i, j, m) == + LET logOk == \/ m.mlastLogTerm > LastTerm(log[i]) + \/ /\ m.mlastLogTerm = LastTerm(log[i]) + /\ m.mlastLogIndex >= LastIndex(log[i]) + grant == /\ m.mterm = currentTerm[i] + /\ logOk + /\ votedFor[i] \in {Nil, j} + IN /\ m.mterm <= currentTerm[i] + /\ \/ grant /\ votedFor' = [votedFor EXCEPT ![i] = j] + \/ ~grant /\ UNCHANGED votedFor + /\ Reply([mtype |-> RequestVoteResponse, + mterm |-> currentTerm[i], + mvoteGranted |-> grant, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + +\* Server i receives a RequestVote response from server j. +\* @type: (Int, Int, RVRESPT) => Bool; +HandleRequestVoteResponse(i, j, m) == + /\ m.mterm = currentTerm[i] + /\ j = i \/ j \in (GetConfig(i) \union GetOutgoingConfig(i) \union GetLearners(i)) + /\ votesResponded' = [votesResponded EXCEPT ![i] = + votesResponded[i] \cup {j}] + /\ \/ /\ m.mvoteGranted + /\ votesGranted' = [votesGranted EXCEPT ![i] = + votesGranted[i] \cup {j}] + \/ /\ ~m.mvoteGranted + /\ UNCHANGED <> + /\ Discard(m) + /\ UNCHANGED <> + +\* @type: (Int, Int, AEREQT, Bool) => Bool; +RejectAppendEntriesRequest(i, j, m, logOk) == + /\ \/ m.mterm < currentTerm[i] + \/ /\ m.mterm = currentTerm[i] + /\ state[i] = Follower + /\ \lnot logOk + \* Rejection uses findConflictByTerm for fast backtracking + /\ LET hintIndex == Min({m.mprevLogIndex, LastIndex(log[i])}) + rejectHint == FindConflictByTerm(i, hintIndex, m.mprevLogTerm) + logTerm == LogTerm(i, rejectHint) + IN Reply([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> FALSE, + mmatchIndex |-> m.mprevLogIndex, + mrejectHint |-> rejectHint, + mlogTerm |-> logTerm, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + +\* @type: (Int, MSG) => Bool; +ReturnToFollowerState(i, m) == + /\ m.mterm = currentTerm[i] + /\ state[i] = Candidate + /\ state' = [state EXCEPT ![i] = Follower] + /\ UNCHANGED <> + +HasNoConflict(i, index, ents) == + /\ index <= LastIndex(log[i]) + 1 + /\ \A k \in 1..Len(ents): index + k - 1 <= LastIndex(log[i]) => LogTerm(i, index+k-1) = ents[k].term + +\* Find first conflicting entry index (0 if none). Reference: log.go findConflict +\* @type: (Int, Int, Seq(ENTRY)) => Int; +FindFirstConflict(i, index, ents) == + LET conflicting == {k \in 1..Len(ents): + /\ index + k - 1 <= LastIndex(log[i]) + /\ LogTerm(i, index + k - 1) /= ents[k].term} + IN IF conflicting = {} THEN 0 ELSE index + Min(conflicting) - 1 + +\* @type: (Int, Int, Int, AEREQT) => Bool; +AppendEntriesAlreadyDone(i, j, index, m) == + /\ \/ index <= commitIndex[i] + \/ /\ index > commitIndex[i] + /\ \/ m.mentries = << >> + \/ /\ m.mentries /= << >> + /\ m.mprevLogIndex + Len(m.mentries) <= LastIndex(log[i]) + /\ HasNoConflict(i, index, m.mentries) + /\ IF index <= commitIndex[i] THEN + IF m.msubtype = "heartbeat" THEN CommitTo(i, m.mcommitIndex) ELSE UNCHANGED commitIndex + ELSE + CommitTo(i, Min({m.mcommitIndex, m.mprevLogIndex+Len(m.mentries)})) + /\ Reply([ mtype |-> AppendEntriesResponse, + msubtype |-> m.msubtype, + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> IF m.msubtype = "heartbeat" \/ index > commitIndex[i] THEN m.mprevLogIndex+Len(m.mentries) ELSE commitIndex[i], + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + +\* @type: (Int, Int, Int, AEREQT) => Bool; +ConflictAppendEntriesRequest(i, j, index, m) == + /\ m.mentries /= << >> + /\ index > commitIndex[i] + /\ ~HasNoConflict(i, index, m.mentries) + \* Find conflict, truncate, and append (Reference: log.go maybeAppend + findConflict) + /\ LET ci == FindFirstConflict(i, index, m.mentries) + entsOffset == ci - index + 1 + newEntries == SubSeq(m.mentries, entsOffset, Len(m.mentries)) + keepUntil == ci - log[i].offset + IN /\ ci > commitIndex[i] + /\ log' = [log EXCEPT ![i].entries = SubSeq(@, 1, keepUntil) \o newEntries] + /\ historyLog' = [historyLog EXCEPT ![i] = SubSeq(@, 1, ci - 1) \o newEntries] + /\ CommitTo(i, Min({m.mcommitIndex, m.mprevLogIndex + Len(m.mentries)})) + /\ Reply([mtype |-> AppendEntriesResponse, + msubtype |-> m.msubtype, + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> m.mprevLogIndex + Len(m.mentries), + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + +\* @type: (Int, Int, Int, AEREQT) => Bool; +NoConflictAppendEntriesRequest(i, j, index, m) == + /\ m.mentries /= << >> + /\ index > commitIndex[i] + /\ HasNoConflict(i, index, m.mentries) + /\ m.mprevLogIndex + Len(m.mentries) > LastIndex(log[i]) + /\ LET newEntries == SubSeq(m.mentries, LastIndex(log[i])-index+2, Len(m.mentries)) + IN /\ log' = [log EXCEPT ![i].entries = @ \o newEntries] + /\ historyLog' = [historyLog EXCEPT ![i] = @ \o newEntries] + /\ CommitTo(i, Min({m.mcommitIndex, m.mprevLogIndex + Len(m.mentries)})) + /\ Reply([mtype |-> AppendEntriesResponse, + msubtype |-> m.msubtype, + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> m.mprevLogIndex + Len(m.mentries), + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + +\* @type: (Int, Int, Bool, AEREQT) => Bool; +AcceptAppendEntriesRequest(i, j, logOk, m) == + \* accept request + /\ m.mterm = currentTerm[i] + /\ state[i] = Follower + /\ logOk + /\ LET index == m.mprevLogIndex + 1 + IN \/ AppendEntriesAlreadyDone(i, j, index, m) + \/ ConflictAppendEntriesRequest(i, j, index, m) + \/ NoConflictAppendEntriesRequest(i, j, index, m) + +\* Server i receives an AppendEntries request from server j with +\* m.mterm <= currentTerm[i]. This just handles m.entries of length 0 or 1, but +\* implementations could safely accept more by treating them the same as +\* multiple independent requests of 1 entry. +\* @type: (Int, Int, AEREQT) => Bool; +HandleAppendEntriesRequest(i, j, m) == + LET logOk == \/ m.mprevLogIndex = 0 + \/ /\ m.mprevLogIndex > 0 + /\ m.mprevLogIndex <= LastIndex(log[i]) + /\ m.mprevLogTerm = LogTerm(i, m.mprevLogIndex) + IN + /\ m.mterm <= currentTerm[i] + /\ \/ RejectAppendEntriesRequest(i, j, m, logOk) + \/ ReturnToFollowerState(i, m) + \/ AcceptAppendEntriesRequest(i, j, logOk, m) + /\ UNCHANGED <> + +\* Server i receives an AppendEntries response from server j. +\* @type: (Int, Int, AERESPT) => Bool; +HandleAppendEntriesResponse(i, j, m) == + /\ m.mterm = currentTerm[i] + /\ m.msubtype /= "heartbeat" \* Heartbeat responses handled by HandleHeartbeatResponse + /\ \/ /\ m.msuccess \* successful + /\ matchIndex' = [matchIndex EXCEPT ![i][j] = Max({@, m.mmatchIndex})] + /\ UNCHANGED <> + /\ FreeInflightsLE(i, j, m.mmatchIndex) + \* State transition logic (Reference: raft.go handleAppendEntriesResponse()) + /\ LET maybeUpdated == m.mmatchIndex > matchIndex[i][j] + alreadyMatched == m.mmatchIndex = matchIndex[i][j] + newMatchIndex == Max({matchIndex[i][j], m.mmatchIndex}) + \* Check if follower caught up enough to resume from snapshot + canResumeFromSnapshot == \/ newMatchIndex + 1 >= log[i].offset + \/ newMatchIndex + 1 >= pendingSnapshot[i][j] + IN CASE \* StateProbe -> StateReplicate + progressState[i][j] = StateProbe + /\ (maybeUpdated \/ alreadyMatched) -> + /\ progressState' = [progressState EXCEPT ![i][j] = StateReplicate] + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = Max({@, m.mmatchIndex + 1})] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ UNCHANGED pendingSnapshot + \* StateSnapshot -> StateReplicate (if caught up) + [] progressState[i][j] = StateSnapshot + /\ maybeUpdated + /\ canResumeFromSnapshot -> + /\ progressState' = [progressState EXCEPT ![i][j] = StateReplicate] + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = m.mmatchIndex + 1] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i][j] = 0] + \* Other: only clear MsgAppFlowPaused if matchIndex actually updated + [] OTHER -> + /\ UNCHANGED <> + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = Max({@, m.mmatchIndex + 1})] + /\ IF maybeUpdated + THEN msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + ELSE UNCHANGED msgAppFlowPaused + \/ /\ \lnot m.msuccess \* MaybeDecrTo (progress.go:226-252) + /\ LET rejected == m.mmatchIndex + matchHint == m.mrejectHint + IN IF progressState[i][j] = StateReplicate + THEN IF rejected <= matchIndex[i][j] + THEN /\ UNCHANGED <> \* stale + ELSE \* transition to Probe + /\ progressState' = [progressState EXCEPT ![i][j] = StateProbe] + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = matchIndex[i][j] + 1] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i][j] = 0] + /\ inflights' = [inflights EXCEPT ![i][j] = {}] + /\ UNCHANGED <> + ELSE IF nextIndex[i][j] - 1 /= rejected + THEN \* stale, just unpause + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ UNCHANGED <> + ELSE \* Leader-side findConflictByTerm optimization + LET leaderMatchIdx == FindConflictByTerm(i, matchHint, m.mlogTerm) + newNext == IF leaderMatchIdx > matchIndex[i][j] + THEN leaderMatchIdx + 1 + ELSE Max({Min({rejected, matchHint + 1}), matchIndex[i][j] + 1}) + IN + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = newNext] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ UNCHANGED <> + /\ Discard(m) + /\ UNCHANGED <> + +\* Server i receives a heartbeat response from server j. Only clears flow control. +HandleHeartbeatResponse(i, j, m) == + /\ m.mterm = currentTerm[i] + /\ m.msubtype = "heartbeat" + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = FALSE] + /\ Discard(m) + /\ UNCHANGED <> + +\* Compacts log of server i up to newStart (exclusive). Reference: storage.go Compact() +CompactLog(i, newStart) == + /\ newStart > log[i].offset + /\ newStart <= applied[i] + 1 + /\ LET compactedEntries == SubSeq(log[i].entries, 1, newStart - log[i].offset) + IN + /\ log' = [log EXCEPT ![i] = [ + offset |-> newStart, + entries |-> SubSeq(@.entries, newStart - @.offset + 1, Len(@.entries)), + snapshotIndex |-> newStart - 1, + snapshotTerm |-> LogTerm(i, newStart - 1) + ]] + /\ durableState' = [durableState EXCEPT ![i] = [ + @ EXCEPT !.snapshotHistory = @ \o compactedEntries, + !.snapshotIndex = newStart - 1, + !.snapshotTerm = LogTerm(i, newStart - 1), + !.entries = SubSeq(log[i].entries, newStart - log[i].offset + 1, Len(log[i].entries)) + ]] + /\ UNCHANGED <> + +\* Server i receives a SnapshotRequest. Reference: raft.go restore() +HandleSnapshotRequest(i, j, m) == + /\ m.mterm <= currentTerm[i] + /\ IF m.mterm < currentTerm[i] THEN + \* Stale term: ignore + /\ Discard(m) + /\ UNCHANGED <> + ELSE IF m.msnapshotIndex <= commitIndex[i] THEN + \* Stale snapshot (already committed) + /\ Reply([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> commitIndex[i], + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + ELSE IF LogTerm(i, m.msnapshotIndex) = m.msnapshotTerm THEN + \* Fast-forward: log already contains this snapshot's index/term + /\ commitIndex' = [commitIndex EXCEPT ![i] = m.msnapshotIndex] + /\ Reply([mtype |-> AppendEntriesResponse, + msubtype |-> "app", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> m.msnapshotIndex, + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + ELSE + \* Actual restore: wipe log and restore config atomically + LET confState == m.mconfState + configIndices == {k \in 1..Len(m.mhistory) : m.mhistory[k].type = ConfigEntry} + lastConfigIdx == IF configIndices /= {} THEN Max(configIndices) ELSE 0 + IN + /\ log' = [log EXCEPT ![i] = [ + offset |-> m.msnapshotIndex + 1, + entries |-> <<>>, + snapshotIndex |-> m.msnapshotIndex, + snapshotTerm |-> m.msnapshotTerm + ]] + /\ historyLog' = [historyLog EXCEPT ![i] = m.mhistory] + /\ commitIndex' = [commitIndex EXCEPT ![i] = m.msnapshotIndex] + \* applied updated later via ApplyEntries action + /\ UNCHANGED applied + /\ config' = [config EXCEPT ![i] = [ + jointConfig |-> <>, + learners |-> confState.learners, + autoLeave |-> confState.autoLeave + ]] + /\ appliedConfigIndex' = [appliedConfigIndex EXCEPT ![i] = lastConfigIdx] + /\ durableState' = [durableState EXCEPT ![i] = [ + @ EXCEPT !.log = m.msnapshotIndex, + !.entries = <<>>, + !.snapshotIndex = m.msnapshotIndex, + !.snapshotTerm = m.msnapshotTerm, + !.snapshotHistory = m.mhistory, + !.commitIndex = m.msnapshotIndex, + !.config = config'[i] + ]] + /\ Reply([mtype |-> AppendEntriesResponse, + msubtype |-> "snapshot", + mterm |-> currentTerm[i], + msuccess |-> TRUE, + mmatchIndex |-> m.msnapshotIndex, + mrejectHint |-> 0, + mlogTerm |-> 0, + msource |-> i, + mdest |-> j], + m) + /\ UNCHANGED <> + +\* Handle ReportUnreachable from application layer +\* Reference: raft.go:1624-1632 +\* Application reports that a peer is unreachable, causing StateReplicate -> StateProbe +\* Reference: tracker/progress.go:121-126 ResetState() clears Inflights +\* @type: (Int, Int) => Bool; +ReportUnreachable(i, j) == + /\ state[i] = Leader + /\ i # j + /\ IF progressState[i][j] = StateReplicate + THEN /\ progressState' = [progressState EXCEPT ![i][j] = StateProbe] + /\ inflights' = [inflights EXCEPT ![i][j] = {}] + ELSE UNCHANGED <> + /\ UNCHANGED <> + +\* Handle ReportSnapshot from application layer. Reference: raft.go:1608-1625 +\* @type: (Int, Int, Bool) => Bool; +ReportSnapshotStatus(i, j, success) == + /\ state[i] = Leader + /\ progressState[i][j] = StateSnapshot + /\ LET oldPendingSnapshot == IF success THEN pendingSnapshot[i][j] ELSE 0 + newNext == Max({matchIndex[i][j] + 1, oldPendingSnapshot + 1}) + IN /\ progressState' = [progressState EXCEPT ![i][j] = StateProbe] + /\ nextIndex' = [nextIndex EXCEPT ![i][j] = newNext] + /\ pendingSnapshot' = [pendingSnapshot EXCEPT ![i][j] = 0] + /\ msgAppFlowPaused' = [msgAppFlowPaused EXCEPT ![i][j] = TRUE] + /\ inflights' = [inflights EXCEPT ![i][j] = {}] + /\ UNCHANGED <> + +\* Any RPC with a newer term causes the recipient to advance its term first. +\* @type: (Int, Int, MSG) => Bool; +UpdateTerm(i, j, m) == + /\ m.mterm > currentTerm[i] + /\ BecomeFollowerOfTerm(i, m.mterm) + \* messages is unchanged so m can be processed further. + /\ UNCHANGED <> + +\* Responses with stale terms are ignored. +\* @type: (Int, Int, MSG) => Bool; +DropStaleResponse(i, j, m) == + /\ m.mterm < currentTerm[i] + /\ Discard(m) + /\ UNCHANGED <> + +\* Drop responses from non-members. Self-directed messages bypass this check. +\* Reference: rawnode.go:123-125 +\* @type: (Int, Int, MSG) => Bool; +DropResponseFromNonMember(i, j, m) == + /\ j /= i + /\ j \notin (GetConfig(i) \union GetOutgoingConfig(i) \union GetLearners(i)) + /\ Discard(m) + /\ UNCHANGED <> + +\* Atomic term update + vote handling (single Step call in raft.go). +UpdateTermAndHandleRequestVote(i, j, m) == + /\ m.mtype = RequestVoteRequest + /\ m.mterm > currentTerm[i] + /\ LET logOk == \/ m.mlastLogTerm > LastTerm(log[i]) + \/ /\ m.mlastLogTerm = LastTerm(log[i]) + /\ m.mlastLogIndex >= LastIndex(log[i]) + grant == logOk \* Term is equal (after update), Vote is Nil (after update) + IN + /\ Reply([mtype |-> RequestVoteResponse, + mterm |-> m.mterm, + mvoteGranted |-> grant, + msource |-> i, + mdest |-> j], + m) + /\ currentTerm' = [currentTerm EXCEPT ![i] = m.mterm] + /\ state' = [state EXCEPT ![i] = Follower] + /\ votedFor' = [votedFor EXCEPT ![i] = IF grant THEN j ELSE Nil] + /\ UNCHANGED <> + +\* Receive a message (with optional FIFO ordering and partition checks). +ReceiveDirect(m) == + LET i == m.mdest + j == m.msource + IN /\ CanCommunicate(j, i) \* Partition check: source and dest must be able to communicate + /\ IsFifoFirst(m) \* FIFO constraint: only receive if first in order + /\ \* Any RPC with a newer term causes the recipient to advance + \* its term first. Responses with stale terms are ignored. + \/ UpdateTermAndHandleRequestVote(i, j, m) + \/ /\ m.mtype /= RequestVoteRequest + /\ UpdateTerm(i, j, m) + \/ /\ m.mtype = RequestVoteRequest + /\ HandleRequestVoteRequest(i, j, m) + \/ /\ m.mtype = RequestVoteResponse + /\ \/ DropStaleResponse(i, j, m) + \/ DropResponseFromNonMember(i, j, m) + \/ HandleRequestVoteResponse(i, j, m) + \/ /\ m.mtype = AppendEntriesRequest + /\ HandleAppendEntriesRequest(i, j, m) + \/ /\ m.mtype = AppendEntriesResponse + /\ \/ DropStaleResponse(i, j, m) + \/ DropResponseFromNonMember(i, j, m) + \/ HandleHeartbeatResponse(i, j, m) + \/ HandleAppendEntriesResponse(i, j, m) + \/ /\ m.mtype = SnapshotRequest + /\ HandleSnapshotRequest(i, j, m) + +Receive(m) == ReceiveDirect(m) + +NextRequestVoteRequest == \E m \in DOMAIN messages : m.mtype = RequestVoteRequest /\ Receive(m) +NextRequestVoteResponse == \E m \in DOMAIN messages : m.mtype = RequestVoteResponse /\ Receive(m) +NextAppendEntriesRequest == \E m \in DOMAIN messages : m.mtype = AppendEntriesRequest /\ Receive(m) +NextAppendEntriesResponse == \E m \in DOMAIN messages : m.mtype = AppendEntriesResponse /\ Receive(m) + +\* End of message handlers. +---- +\* Network state transitions + +\* The network duplicates a message +\* @type: MSG => Bool; +DuplicateMessage(m) == + /\ m \in DOMAIN messages + /\ messages' = WithMessage(m, messages) + /\ UNCHANGED <> + +\* The network drops a message +\* @type: MSG => Bool; +DropMessage(m) == + \* Do not drop loopback messages + \* /\ m.msource /= m.mdest + /\ Discard(m) + /\ UNCHANGED <> + +\* Network Partition Actions + +\* Create a network partition, dropping all cross-partition messages. +CreatePartition(partitionAssignment) == + /\ ~IsPartitioned + /\ \A i \in Server : partitionAssignment[i] \in 1..MaxPartitions + \* /\ Cardinality({partitionAssignment[i] : i \in Server}) >= 2 + /\ partitions' = partitionAssignment + /\ messages' = messages (-) SetToBag(CrossPartitionMessages) + /\ pendingMessages' = pendingMessages (-) SetToBag(CrossPartitionPendingMessages) + /\ UNCHANGED <> + +\* Heal the network partition. +HealPartition == + /\ IsPartitioned + /\ partitions' = [i \in Server |-> 0] + /\ UNCHANGED <> + +---- + +\* Defines how the variables may transition. +NextAsync == + \/ \E i,j \in Server : RequestVote(i, j) + \/ \E i \in Server : BecomeLeader(i) + \/ \E i \in Server: ClientRequest(i, 0) + \/ \E i \in Server: ClientRequestAndSend(i, 0) + \/ \E i \in Server : AdvanceCommitIndex(i) + \/ \E i,j \in Server : \E b,e \in matchIndex[i][j]+1..LastIndex(log[i])+1 : AppendEntries(i, j, <>) + \/ \E i \in Server : AppendEntriesToSelf(i) + \/ \E i,j \in Server : Heartbeat(i, j) + \/ \E i,j \in Server : SendSnapshot(i, j) + \/ \E i \in Server : IF applied[i] < commitIndex[i] THEN ApplyEntries(i, commitIndex[i]) ELSE FALSE + \/ \E i \in Server : IF log[i].offset < applied[i] THEN CompactLog(i, applied[i]) ELSE FALSE + \/ \E m \in DOMAIN messages : Receive(m) + \/ \E i \in Server : Timeout(i) + \/ \E i \in Server : Ready(i) + \/ \E i \in Server : StepDownToFollower(i) + \/ \E i,j \in Server : ReportUnreachable(i, j) + \/ \E i,j \in Server : ReportSnapshotStatus(i, j, TRUE) + \/ \E i,j \in Server : ReportSnapshotStatus(i, j, FALSE) + +NextCrash == \E i \in Server : Restart(i) + +NextAsyncCrash == + \/ NextAsync + \/ NextCrash + +NextUnreliable == + \* Only duplicate once + \/ \E m \in DOMAIN messages : + /\ messages[m] = 1 + /\ DuplicateMessage(m) + \* Only drop if it makes a difference + \/ \E m \in DOMAIN messages : + /\ messages[m] = 1 + /\ DropMessage(m) + +\* Most pessimistic network model +Next == \/ NextAsync + \/ NextCrash + \/ NextUnreliable + +\* Membership changes +\* Note: AddNewServer, AddLearner, DeleteServer are removed from NextDynamic. +\* They bypass ChangeConf constraints and can cause QuorumLogInv violations. +\* Use ChangeConf with enterJoint parameter instead. +NextDynamic == + \/ Next + \/ \E i \in Server : ChangeConf(i) + \/ \E i \in Server : ChangeConfAndSend(i) + \/ \E i \in Server : ApplySimpleConfChange(i) + \/ \E i \in Server : ProposeLeaveJoint(i) + +\* The specification must start with the initial state and transition according +\* to Next. +Spec == Init /\ [][Next]_vars + +(***************************************************************************) +(* The main safety properties are below *) +(***************************************************************************) +---- + +ASSUME DistinctRoles == /\ Leader /= Candidate + /\ Candidate /= Follower + /\ Follower /= Leader + +ASSUME DistinctMessageTypes == /\ RequestVoteRequest /= AppendEntriesRequest + /\ RequestVoteRequest /= RequestVoteResponse + /\ RequestVoteRequest /= AppendEntriesResponse + /\ AppendEntriesRequest /= RequestVoteResponse + /\ AppendEntriesRequest /= AppendEntriesResponse + /\ RequestVoteResponse /= AppendEntriesResponse + /\ SnapshotRequest /= RequestVoteRequest + /\ SnapshotRequest /= AppendEntriesRequest + /\ SnapshotRequest /= RequestVoteResponse + /\ SnapshotRequest /= AppendEntriesResponse + /\ SnapshotResponse /= RequestVoteRequest + /\ SnapshotResponse /= AppendEntriesRequest + /\ SnapshotResponse /= RequestVoteResponse + /\ SnapshotResponse /= AppendEntriesResponse + /\ SnapshotRequest /= SnapshotResponse + +---- +\* Correctness invariants + +\* The prefix of the log of server i that has been committed +Committed(i) == SubSeq(historyLog[i],1,commitIndex[i]) + +\* @type: MSG => Bool; +MessageTermsLtCurrentTerm(m) == + m.mterm <= currentTerm[m.msource] + +\* Committed log entries should never conflict between servers +LogInv == + \A i, j \in Server : i /= j => + \/ IsPrefix(Committed(i),Committed(j)) + \/ IsPrefix(Committed(j),Committed(i)) + + +\* At most one leader per term +MoreThanOneLeaderInv == + \A i,j \in Server : + (/\ currentTerm[i] = currentTerm[j] + /\ state[i] = Leader + /\ state[j] = Leader) + => i = j + +\* Every (index, term) pair determines a log prefix +LogMatchingInv == + \A i, j \in Server : i /= j => + LET minLen == Min({Len(historyLog[i]), Len(historyLog[j])}) + \* Find all indices where terms match + matchingTerms == {n \in 1..minLen : historyLog[i][n].term = historyLog[j][n].term} + IN matchingTerms /= {} => + \* Only check the maximum matching index + SubSeq(historyLog[i], 1, Max(matchingTerms)) = SubSeq(historyLog[j], 1, Max(matchingTerms)) + +\* All committed entries exist in at least one server per quorum. +\* In joint config, checks outgoing config's quorum. +\* Only checks servers with up-to-date config. +QuorumLogInv == + \A i \in Server : + \* Find config entries within the committed range + LET configIndicesInCommitted == {k \in 1..commitIndex[i] : + k <= Len(historyLog[i]) /\ historyLog[i][k].type = ConfigEntry} + \* Check if server's config is up-to-date (applied all committed config entries) + configUpToDate == configIndicesInCommitted = {} \/ + appliedConfigIndex[i] >= Max(configIndicesInCommitted) + \* In joint config, use outgoing config for quorum check + \* because incoming config hasn't taken effect yet (LeaveJoint not committed) + effectiveConfig == IF IsJointConfig(i) THEN GetOutgoingConfig(i) ELSE GetConfig(i) + IN + \* Only check servers with up-to-date config + configUpToDate => + \A S \in Quorum(effectiveConfig) : + \E j \in S : IsPrefix(Committed(i), historyLog[j]) + +\* A more up-to-date server has all committed entries of less up-to-date servers +MoreUpToDateCorrectInv == + \A i, j \in Server : i /= j => + ((\/ LastTerm(log[i]) > LastTerm(log[j]) + \/ /\ LastTerm(log[i]) = LastTerm(log[j]) + /\ LastIndex(log[i]) >= LastIndex(log[j])) => + IsPrefix(Committed(j), historyLog[i])) + +\* Leader must contain all entries committed in previous terms by any server. + +\* Helper: prefix of committed log where term <= t +CommittedTermPrefix(j, t) == + LET committed == Committed(j) + validIndices == {k \in 1..Len(committed) : committed[k].term <= t} + IN IF validIndices = {} THEN <<>> + ELSE SubSeq(committed, 1, Max(validIndices)) + +LeaderCompletenessInv == + \A i \in Server : + state[i] = Leader => + \A j \in Server : i /= j => + IsPrefix(CommittedTermPrefix(j, currentTerm[i]), historyLog[i]) + +\* Any entry committed by leader shall be persisted already +CommittedIsDurableInv == + \A i \in Server : + state[i] = Leader => commitIndex[i] <= durableState[i].log + + + +ProbeLimitInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateProbe) + => InflightsCount(i, j) <= 1 + +\* In StateProbe, at most 1 AppendEntries message in the network +ProbeNetworkMessageLimitInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ i /= j /\ progressState[i][j] = StateProbe) => + LET msgsInNetwork == {m \in DOMAIN messages : + m.mtype = AppendEntriesRequest /\ + m.msource = i /\ + m.mdest = j} + totalCount == FoldSet(LAMBDA m, acc: acc + messages[m], 0, msgsInNetwork) + IN totalCount <= 1 + +\* In StateReplicate, paused only when inflights full +ReplicatePauseInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateReplicate /\ msgAppFlowPaused[i][j]) + => InflightsFull(i, j) + + +\* In StateSnapshot, inflights must be empty +SnapshotInflightsInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateSnapshot) + => InflightsCount(i, j) = 0 + +\* Inflight indices must be <= LastIndex(log) +InflightsLogIndexInv == + \A i \in Server : \A j \in Server : + state[i] = Leader => + \A idx \in inflights[i][j] : idx <= LastIndex(log[i]) + +\* Inflight indices must be > matchIndex +InflightsMatchIndexInv == + \A i \in Server : \A j \in Server : + state[i] = Leader => + \A idx \in inflights[i][j] : idx > matchIndex[i][j] + + +ProgressStateTypeInv == + \A i, j \in Server: + progressState[i][j] \in {StateProbe, StateReplicate, StateSnapshot} + +\* Reference: inflights.go:66-68 Add() panics on full inflights +InflightsInv == + \A i, j \in Server: + InflightsCount(i, j) <= MaxInflightMsgs + +SnapshotPendingInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateSnapshot) + => pendingSnapshot[i][j] > 0 + +NoPendingSnapshotInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] /= StateSnapshot) + => pendingSnapshot[i][j] = 0 + +LeaderSelfReplicateInv == + \A i \in Server : + state[i] = Leader => progressState[i][i] = StateReplicate + +\* Comprehensive StateSnapshot consistency check +SnapshotStateInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateSnapshot) => + /\ InflightsCount(i, j) = 0 \* Inflights must be empty + /\ pendingSnapshot[i][j] > 0 \* Must have pending snapshot + /\ pendingSnapshot[i][j] <= LastIndex(log[i]) \* Snapshot index must be valid + +InflightsMonotonicInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ inflights[i][j] # {}) => + LET maxIdx == Max(inflights[i][j]) + minIdx == Min(inflights[i][j]) + IN + maxIdx >= minIdx \* Trivial sanity check + +\* Reference: progress.go:37 "0 <= Match < Next" +MatchIndexLessThanLogInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ j /= i) => + matchIndex[i][j] <= LastIndex(log[i]) + +MatchIndexNonNegativeInv == + \A i \in Server : \A j \in Server : + matchIndex[i][j] >= 0 + +\* Inflight indices in (Match, Next) interval +InflightsAboveMatchInv == + \A i \in Server : \A j \in Server : + state[i] = Leader => + \A idx \in inflights[i][j] : idx > matchIndex[i][j] + +\* Match < Next (fundamental Progress invariant) +\* Reference: progress.go:37 +MatchIndexLessThanNextInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ j /= i) => + matchIndex[i][j] < nextIndex[i][j] + + +\* In StateReplicate: not paused => inflights not full +MsgAppFlowPausedConsistencyInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateReplicate + /\ ~msgAppFlowPaused[i][j]) => + ~InflightsFull(i, j) + +ProbeOneInflightMaxInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateProbe) => + InflightsCount(i, j) <= 1 + +SnapshotNoInflightsStrictInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateSnapshot) => + InflightsCount(i, j) = 0 + +ProgressSafety == + /\ ProbeLimitInv + /\ ProbeNetworkMessageLimitInv + /\ ReplicatePauseInv + /\ SnapshotInflightsInv + /\ InflightsLogIndexInv + /\ InflightsMatchIndexInv + /\ ProgressStateTypeInv + /\ InflightsInv + /\ SnapshotPendingInv + /\ NoPendingSnapshotInv + /\ LeaderSelfReplicateInv + /\ SnapshotStateInv + /\ InflightsMonotonicInv + /\ MatchIndexLessThanLogInv + /\ MatchIndexNonNegativeInv + /\ InflightsAboveMatchInv + /\ MatchIndexLessThanNextInv + /\ MsgAppFlowPausedConsistencyInv + /\ ProbeOneInflightMaxInv + /\ SnapshotNoInflightsStrictInv + + + +AllMessageTermsValid == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + MessageTermsLtCurrentTerm(m) + +MessageIndexValidInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + (m.mtype = AppendEntriesRequest) => + /\ m.mprevLogIndex >= 0 + /\ m.mcommitIndex >= 0 + +\* All nodes agree on committed entries +StateMachineConsistency == + \A i, j \in Server : i /= j => + LET minCommit == Min({commitIndex[i], commitIndex[j]}) + IN SubSeq(historyLog[i], 1, minCommit) = SubSeq(historyLog[j], 1, minCommit) + +CommitIndexBoundInv == + \A i \in Server : + commitIndex[i] <= LastIndex(log[i]) + +\* Log terms are monotonically non-decreasing +LogTermMonotonic == + \A i \in Server : + \A idx \in 1..(LastIndex(log[i]) - 1) : + LogTerm(i, idx) <= LogTerm(i, idx + 1) + +CommittedEntriesTermInv == + \A i \in Server : + \A idx \in 1..commitIndex[i] : + historyLog[i][idx].term > 0 + +PendingConfigBoundInv == + \A i \in Server : + state[i] = Leader => + pendingConfChangeIndex[i] <= LastIndex(log[i]) + +LeaderLogLengthInv == + \A i \in Server : + state[i] = Leader => + commitIndex[i] <= LastIndex(log[i]) + +\* currentTerm >= all log entry terms +CurrentTermAtLeastLogTerm == + \A i \in Server : + \* Check snapshot term + /\ log[i].snapshotTerm <= currentTerm[i] + \* Check only the last entry (highest term due to monotonicity) + /\ Len(log[i].entries) > 0 => + log[i].entries[Len(log[i].entries)].term <= currentTerm[i] + + +CandidateVotedForSelfInv == + \A i \in Server : + state[i] = Candidate => + votedFor[i] = i + +\* Durable state <= volatile state +DurableStateConsistency == + \A i \in Server : + /\ durableState[i].currentTerm <= currentTerm[i] + /\ durableState[i].commitIndex <= commitIndex[i] + +MessageEndpointsValidInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + /\ m.msource \in Server + /\ m.mdest \in Server + +LeaderDurableTermInv == + \A i \in Server : + state[i] = Leader => + durableState[i].currentTerm = currentTerm[i] + +AdditionalSafety == + /\ AllMessageTermsValid + /\ MessageIndexValidInv + /\ StateMachineConsistency + /\ CommitIndexBoundInv + /\ CommittedEntriesTermInv + /\ PendingConfigBoundInv + /\ LeaderLogLengthInv + /\ CurrentTermAtLeastLogTerm + /\ CandidateVotedForSelfInv + /\ DurableStateConsistency + /\ MessageEndpointsValidInv + /\ LeaderDurableTermInv + +LogOffsetMinInv == + \A i \in Server : + log[i].offset >= 1 + +\* snapshotIndex = offset - 1 +SnapshotOffsetConsistencyInv == + \A i \in Server : + log[i].snapshotIndex = log[i].offset - 1 + +SnapshotTermValidInv == + \A i \in Server : + log[i].snapshotIndex > 0 => log[i].snapshotTerm > 0 + +SnapshotTermBoundInv == + \A i \in Server : + log[i].snapshotTerm <= currentTerm[i] + +HistoryLogLengthInv == + \A i \in Server : + Len(historyLog[i]) = LastIndex(log[i]) + +LogStructureInv == + /\ LogOffsetMinInv + /\ SnapshotOffsetConsistencyInv + /\ SnapshotTermValidInv + /\ SnapshotTermBoundInv + /\ HistoryLogLengthInv + + +JointConfigNonEmptyInv == + \A i \in Server : + IsJointConfig(i) => + /\ GetConfig(i) /= {} + /\ GetOutgoingConfig(i) /= {} + +SingleConfigOutgoingEmptyInv == + \A i \in Server : + ~IsJointConfig(i) => GetOutgoingConfig(i) = {} + +\* Reference: tracker/tracker.go:37-41 +LearnersVotersDisjointInv == + \A i \in Server : + GetLearners(i) \cap (GetConfig(i) \union GetOutgoingConfig(i)) = {} + +\* Initialized servers must have at least one voter +ConfigNonEmptyInv == + \A i \in Server : + LET configIndices == {k \in 1..Len(historyLog[i]) : historyLog[i][k].type = ConfigEntry} + lastConfigIdx == IF configIndices /= {} THEN Max(configIndices) ELSE 0 + \* Config is considered applied if no config entries exist or appliedConfigIndex >= last config + configApplied == lastConfigIdx = 0 \/ appliedConfigIndex[i] >= lastConfigIdx + IN + (LastIndex(log[i]) > 0 /\ configApplied) => GetConfig(i) /= {} + +ConfigurationInv == + /\ JointConfigNonEmptyInv + /\ SingleConfigOutgoingEmptyInv + /\ LearnersVotersDisjointInv + /\ ConfigNonEmptyInv + + +SnapshotMsgIndexValidInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = SnapshotRequest => + m.msnapshotIndex <= applied[m.msource] + +SnapshotMsgTermValidInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = SnapshotRequest => + /\ m.msnapshotTerm > 0 + /\ m.msnapshotTerm <= m.mterm + +AppendEntriesPrevIndexNonNegInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = AppendEntriesRequest => + m.mprevLogIndex >= 0 + +AppendEntriesCommitBoundInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = AppendEntriesRequest => + m.mcommitIndex >= 0 + +VoteRequestLogIndexNonNegInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = RequestVoteRequest => + m.mlastLogIndex >= 0 + +VoteRequestLogTermNonNegInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = RequestVoteRequest => + m.mlastLogTerm >= 0 + +MessageContentInv == + /\ SnapshotMsgIndexValidInv + /\ SnapshotMsgTermValidInv + /\ AppendEntriesPrevIndexNonNegInv + /\ AppendEntriesCommitBoundInv + /\ VoteRequestLogIndexNonNegInv + /\ VoteRequestLogTermNonNegInv + +\* Only StateReplicate can have non-empty inflights +InflightsOnlyInReplicateInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] /= StateReplicate) => + inflights[i][j] = {} + +InflightsBelowNextInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ inflights[i][j] /= {}) => + \A idx \in inflights[i][j] : idx < nextIndex[i][j] + +InflightsRefinedInv == + /\ InflightsOnlyInReplicateInv + /\ InflightsBelowNextInv + +SnapshotCommitConsistencyInv == + \A i \in Server : + log[i].snapshotIndex <= commitIndex[i] + + +\* Pending config change index must be within valid log bounds +PendingConfIndexValidInv == + \A i \in Server : + (state[i] = Leader /\ pendingConfChangeIndex[i] > applied[i]) => + /\ pendingConfChangeIndex[i] <= LastIndex(log[i]) + /\ pendingConfChangeIndex[i] >= log[i].offset + + +NextIndexPositiveInv == + \A i \in Server : \A j \in Server : + state[i] = Leader => nextIndex[i][j] >= 1 + +NextIndexBoundInv == + \A i \in Server : \A j \in Server : + state[i] = Leader => nextIndex[i][j] <= LastIndex(log[i]) + 1 + +MatchIndexBoundInv == + \A i \in Server : \A j \in Server : + state[i] = Leader => matchIndex[i][j] <= LastIndex(log[i]) + +PendingSnapshotBoundInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ pendingSnapshot[i][j] > 0) => + pendingSnapshot[i][j] <= LastIndex(log[i]) + + +AppendEntriesTermConsistentInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + (m.mtype = AppendEntriesRequest /\ Len(m.mentries) > 0) => + \A k \in 1..Len(m.mentries) : + m.mentries[k].term <= m.mterm + +SnapshotMsgIndexPositiveInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = SnapshotRequest => + m.msnapshotIndex > 0 + +ResponseTermPositiveInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + (m.mtype = RequestVoteResponse \/ m.mtype = AppendEntriesResponse) => + m.mterm > 0 + + +TermPositiveInv == + \A i \in Server : + currentTerm[i] >= 0 + +LeaderTermPositiveInv == + \A i \in Server : + state[i] = Leader => currentTerm[i] > 0 + +CandidateTermPositiveInv == + \A i \in Server : + state[i] = Candidate => currentTerm[i] > 0 + +\* votesResponded (excluding self) subset of config +VotesRespondedSubsetInv == + \A i \in Server : + state[i] = Candidate => + (votesResponded[i] \ {i}) \subseteq (GetConfig(i) \union GetOutgoingConfig(i) \union GetLearners(i)) + +VotesGrantedSubsetInv == + \A i \in Server : + state[i] = Candidate => + votesGranted[i] \subseteq votesResponded[i] + + +AdditionalSnapshotInv == + /\ SnapshotCommitConsistencyInv + +AdditionalConfigInv == + /\ PendingConfIndexValidInv + +AdditionalProgressInv == + /\ NextIndexPositiveInv + /\ NextIndexBoundInv + /\ MatchIndexBoundInv + /\ PendingSnapshotBoundInv + +AdditionalMessageInv == + /\ AppendEntriesTermConsistentInv + /\ SnapshotMsgIndexPositiveInv + /\ ResponseTermPositiveInv + +TermAndVoteInv == + /\ TermPositiveInv + /\ LeaderTermPositiveInv + /\ CandidateTermPositiveInv + /\ VotesRespondedSubsetInv + /\ VotesGrantedSubsetInv + + +NewInvariants == + /\ LogStructureInv + /\ ConfigurationInv + /\ MessageContentInv + /\ InflightsRefinedInv + /\ AdditionalSnapshotInv + /\ AdditionalConfigInv + /\ AdditionalProgressInv + /\ AdditionalMessageInv + /\ TermAndVoteInv + +\* Bug 76f1249: prevLogTerm must not be 0 when prevLogIndex > 0 + +AppendEntriesPrevLogTermValidInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + (m.mtype = AppendEntriesRequest /\ m.mprevLogIndex > 0) => + m.mprevLogTerm > 0 + +\* Bug bd3c759: at most one pending leave-joint entry in uncommitted log + +SinglePendingLeaveJointInv == + \A i \in Server : + (state[i] = Leader /\ IsJointConfig(i) /\ config[i].autoLeave) => + LET \* Find all uncommitted leave-joint entries + uncommittedLeaveJoints == {k \in (commitIndex[i]+1)..LastIndex(log[i]) : + /\ IsAvailable(i, k) + /\ LogEntry(i, k).type = ConfigEntry + /\ "leaveJoint" \in DOMAIN LogEntry(i, k).value + /\ LogEntry(i, k).value.leaveJoint = TRUE} + IN Cardinality(uncommittedLeaveJoints) <= 1 + +\* pendingConfIndex should cover pending leave-joint entries +PendingConfIndexAutoLeaveInv == + \A i \in Server : + (state[i] = Leader /\ IsJointConfig(i)) => + LET leaveJointIndices == {k \in (commitIndex[i]+1)..LastIndex(log[i]) : + /\ IsAvailable(i, k) + /\ LogEntry(i, k).type = ConfigEntry + /\ "leaveJoint" \in DOMAIN LogEntry(i, k).value + /\ LogEntry(i, k).value.leaveJoint = TRUE} + IN leaveJointIndices /= {} => + pendingConfChangeIndex[i] >= Min(leaveJointIndices) + +\* Bug #12136: joint state must not get stuck +JointStateAutoLeavePossibleInv == + \A i \in Server : + (state[i] = Leader /\ IsJointConfig(i) /\ config[i].autoLeave = TRUE) => + ((applied[i] >= pendingConfChangeIndex[i]) => + pendingConfChangeIndex[i] < LastIndex(log[i])) + +\* Bug #7280: config must be applied before election +ElectionConfigAppliedInv == + \A i \in Server : + state[i] = Candidate => + LET configIndicesInCommitted == {k \in 1..commitIndex[i] : + k <= Len(historyLog[i]) /\ historyLog[i][k].type = ConfigEntry} + IN configIndicesInCommitted = {} \/ appliedConfigIndex[i] >= Max(configIndicesInCommitted) + +\* Bug #124: matchIndex <= pendingSnapshot in StateSnapshot +SnapshotTransitionCorrectInv == + \A i, j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateSnapshot) => + matchIndex[i][j] <= pendingSnapshot[i][j] + + +BugDetectionInv == + /\ AppendEntriesPrevLogTermValidInv + /\ SinglePendingLeaveJointInv + /\ PendingConfIndexAutoLeaveInv + /\ JointStateAutoLeavePossibleInv + /\ ElectionConfigAppliedInv + /\ SnapshotTransitionCorrectInv + + +\* Reference: log.go:46 "applied <= committed" +AppliedBoundInv == + \A i \in Server : + applied[i] <= commitIndex[i] + +\* Reference: Issue #17081 +MessageDestinationValidInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + /\ m.mdest \in Server + /\ m.msource \in Server + +\* snapshotIndex <= applied +SnapshotAppliedConsistencyInv == + \A i \in Server : + log[i].snapshotIndex <= applied[i] + +AppliedConfigBoundInv == + \A i \in Server : + appliedConfigIndex[i] <= commitIndex[i] + +CriticalInv == + /\ AppliedBoundInv + /\ CommitIndexBoundInv + /\ MessageDestinationValidInv + +HighPriorityInv == + /\ SnapshotAppliedConsistencyInv + +DetailedInv == + /\ AppliedConfigBoundInv + + +\* Reference: progress.go:40 "In StateSnapshot, Next == PendingSnapshot + 1" +StateSnapshotNextInv == + \A i \in Server : \A j \in Server : + (state[i] = Leader /\ progressState[i][j] = StateSnapshot) => + nextIndex[i][j] = pendingSnapshot[i][j] + 1 + +CodeDerivedInv == + /\ StateSnapshotNextInv + +\* votedFor must be a valid server +VotedForConsistencyInv == + \A i \in Server : + votedFor[i] /= Nil => + \* If votedFor is set, it should be a valid server + votedFor[i] \in Server + +\* Entries with leader's current term must exist in leader's log +LeaderCurrentTermEntriesInv == + \A leader \in Server : + state[leader] = Leader => + \A other \in Server : + \A idx \in log[other].offset..(LastIndex(log[other])) : + (IsAvailable(other, idx) /\ LogEntry(other, idx).term = currentTerm[leader]) => + \* Entry must exist in leader's log + /\ idx <= LastIndex(log[leader]) + /\ IsAvailable(leader, idx) + /\ LogEntry(leader, idx).term = currentTerm[leader] + +RequestVoteTermBoundInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = RequestVoteRequest => + \* The message term should be positive + /\ m.mterm > 0 + \* Log term in request should be <= message term + /\ m.mlastLogTerm <= m.mterm + +AppendEntriesResponseTermInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = AppendEntriesResponse => + \* Response term is positive + m.mterm > 0 + +SnapshotResponseTermValidInv == + \A m \in DOMAIN messages \union DOMAIN pendingMessages : + m.mtype = SnapshotResponse => + m.mterm > 0 + +VerdiRaftInspiredInv == + /\ VotedForConsistencyInv + /\ LeaderCurrentTermEntriesInv + /\ RequestVoteTermBoundInv + /\ AppendEntriesResponseTermInv + /\ SnapshotResponseTermValidInv + + +\* Log entries must have term > 0 +TermNeverZeroInLogInv == + \A i \in Server : + \A idx \in log[i].offset..LastIndex(log[i]) : + LogEntry(i, idx).term > 0 + +GitHistoryBugPreventionInv == + /\ TermNeverZeroInLogInv + +=============================================================================== \ No newline at end of file