1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
// Copyright 2015 The etcd Authors
// Copyright 2026 Leo Cheng
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
///|
/// A best guess at where our log stops matching another log whose only known
/// point is `(index, term)` (Raft §5.3, `findConflictByTerm`). Returns the
/// greatest `i <= index` whose term is `<= term` (or is unknown because that
/// index is compacted), together with that term. Both the follower (building a
/// reject hint) and the leader (jumping back on that hint) use it, so a whole
/// run of mismatched terms is skipped in one retry instead of one index at a
/// time.
pub fn Node::find_conflict_by_term(
self : Node,
index : UInt64,
term : UInt64,
) -> (UInt64, UInt64) {
let mut i = index
while i > 0 {
let our = self.term_at(i)
// term_at yields 0 for a compacted/out-of-range index, which `<= term`
// treats as a possible match — exactly etcd's "unknown term" case.
if our <= term {
return (i, our)
}
i = i - 1
}
(0, 0)
}
///|
/// Handle an AppendEntries RPC (Raft §5.3). This performs the log-matching
/// consistency check, stores the entries while truncating any conflicting
/// suffix, and advances the commit index. An empty `entries` acts as the
/// leader's heartbeat.
pub fn Node::handle_append_entries(
self : Node,
args : AppendEntriesArgs,
) -> AppendEntriesReply {
if args.term < self.current_term {
// etcd answers a stale MsgApp with a bare MsgAppResp (raft.go:1157): no
// reject hint and `Index` unset, its only purpose being to reveal our term
// so the superseded leader steps down. `reject_index` is therefore 0.
return {
term: self.current_term,
success: false,
match_index: 0,
conflict_index: 0,
conflict_term: 0,
reject_index: 0,
}
}
// A legitimate leader whose term is at least ours: adopt the term and step
// down to follower for it.
if args.term > self.current_term {
self.become_follower(args.term)
} else {
self.role = Follower
}
// The prefix at or below our commit index is immutable and, by the Log
// Matching property, already agrees with any legitimate leader. Accept it
// outright and report our commit index, so the leader re-anchors there
// instead of probing into committed history (etcd).
if args.prev_log_index < self.commit_index {
return {
term: self.current_term,
success: true,
match_index: self.commit_index,
conflict_index: 0,
conflict_term: 0,
reject_index: 0,
}
}
// Log matching: accept only if our log holds an entry at `prev_log_index`
// carrying `prev_log_term`.
if args.prev_log_index <= self.last_log_index() &&
self.term_at(args.prev_log_index) == args.prev_log_term {
let last_new = self.store_entries(args.prev_log_index, args.entries)
if args.leader_commit > self.commit_index {
self.commit_index = if args.leader_commit < last_new {
args.leader_commit
} else {
last_new
}
}
return {
term: self.current_term,
success: true,
match_index: last_new,
conflict_index: 0,
conflict_term: 0,
reject_index: 0,
}
}
// Rejection: hint the leader with the highest (index, term) at or below the
// probe whose term does not exceed the leader's, so it can skip our divergent
// tail in one jump.
let hint = if args.prev_log_index < self.last_log_index() {
args.prev_log_index
} else {
self.last_log_index()
}
let (ci, ct) = self.find_conflict_by_term(hint, args.prev_log_term)
// etcd echoes the rejected probe point back as `MsgAppResp.Index`
// (raft.go:1828), which the leader feeds to `MaybeDecrTo` as `rejected`.
{
term: self.current_term,
success: false,
match_index: 0,
conflict_index: ci,
conflict_term: ct,
reject_index: args.prev_log_index,
}
}
///|
/// Store `entries` immediately after `prev_index`. Where an incoming entry
/// conflicts with an existing one (same index, different term) the local log
/// is truncated at that point before appending. Matching prefixes are left
/// untouched, so applying the same request twice is harmless. Returns the
/// index of the last entry the request covers.
fn Node::store_entries(
self : Node,
prev_index : UInt64,
entries : Array[Entry],
) -> UInt64 {
let mut idx = prev_index
let mut i = 0
while i < entries.length() {
idx = idx + 1
let entry = entries[i]
if idx > self.last_log_index() {
self.log.push(entry)
} else if self.term_at(idx) != entry.term {
// B5: a conflict at or below the commit index would delete a committed
// entry, violating State-Machine Safety. The log-matching check and the
// stale-term guard make this unreachable for a correct leader; assert it
// so a regression surfaces loudly rather than corrupting the log.
if idx <= self.commit_index {
abort("append conflicts with committed entry")
}
self.truncate_from(idx)
self.log.push(entry)
}
i = i + 1
}
idx
}
///|
/// Delete every entry from `index` (1-based) to the end of the log.
fn Node::truncate_from(self : Node, index : UInt64) -> Unit {
let keep = (index - self.snapshot_index - 1).to_int()
while self.log.length() > keep {
self.log.pop() |> ignore
}
}