Skip to main content

raft_rust/raft/
log.rs

1// 区间边界类型:扫描日志时构造半开/闭区间
2use std::ops::{Bound, RangeBounds};
3
4// 序列化/反序列化:日志条目与元数据落盘依赖 bincode + serde
5use serde::{Deserialize, Serialize};
6
7// 成员配置变更条目类型(联合共识 / 单一配置)
8use super::membership::MembershipEntry;
9// 节点 ID 与任期号,日志条目与投票元数据会用到
10use super::{NodeID, Term};
11// 统一错误类型,解码/存储失败向上返回
12use crate::error::Result;
13// 底层键值存储引擎抽象(可插拔)
14use crate::storage;
15
16/// 日志索引(条目位置)。从 1 开始;0 表示无索引。
17pub type Index = u64;
18
19/// Bincode 标准配置,用于日志条目与元数据的序列化。
20const BINCODE: bincode::config::Configuration = bincode::config::standard();
21
22/// 用 bincode 序列化值。
23fn encode_value<T: Serialize>(value: &T) -> Vec<u8> {
24    // 序列化失败视为编程错误:Raft 内部结构必须可编码
25    bincode::serde::encode_to_vec(value, BINCODE).expect("value must be serializable")
26}
27
28/// 用 bincode 反序列化值。
29fn decode_value<'de, T: Deserialize<'de>>(bytes: &'de [u8]) -> Result<T> {
30    // 只取解码结果本体,忽略 bincode 返回的已消费字节数
31    Ok(bincode::serde::borrow_decode_from_slice(bytes, BINCODE)?.0)
32}
33
34/// 包含状态机命令的日志条目。
35#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
36// 日志条目:index/term + 命令或成员变更(互斥)
37pub struct Entry {
38    /// 条目索引。
39    ///
40    /// 编码值里其实可以省略索引(键里也有),但为简单起见仍保留。
41    pub index: Index,
42    /// 条目被加入时的任期。
43    pub term: Term,
44    /// 状态机命令。None(noop)用于领导者选举时提交旧条目,见 Raft 论文 5.4.2 节。
45    /// 与 [`Self::membership`] 互斥:成员变更条目的 command 应为 None。
46    pub command: Option<Vec<u8>>,
47    /// 集群成员配置变更(联合共识 / 单一配置)。追加到日志后立即生效。
48    #[serde(default)]
49    // 可选成员配置;与 command 互斥,追加后立即影响投票集
50    pub membership: Option<MembershipEntry>,
51}
52
53// 条目编解码:与 Key/Engine 之间的字节契约
54impl Entry {
55    // 将本条目编码为可持久化的字节
56    fn encode(&self) -> Vec<u8> {
57        // 复用统一 bincode 编码,保证与存储读写格式一致
58        encode_value(self)
59    }
60
61    // 从存储字节还原日志条目
62    fn decode(bytes: &[u8]) -> Result<Self> {
63        // 解码失败向上传播,避免静默损坏日志
64        decode_value(bytes)
65    }
66}
67
68/// 日志存储键。
69///
70/// 编码为固定前缀 + 可选的大端索引,保证 `Entry(i)` 按索引字典序排列,
71/// 且全部 Entry 键排在元数据键之前:
72/// * `Entry(index)` → `[0x00] ‖ index.to_be_bytes()`
73/// * `TermVote`     → `[0x01]`
74/// * `CommitIndex`  → `[0x02]`
75/// * `SnapshotMeta` → `[0x03]`  (last_included_index, last_included_term)
76/// * `SnapshotData` → `[0x04]`  状态机快照字节
77#[derive(Clone, Debug, PartialEq)]
78// 引擎键空间:条目按索引有序,元数据键排在条目之后
79pub enum Key {
80    /// 日志条目,保存任期与命令。
81    Entry(Index),
82    /// 保存当前任期与投票(若有)。
83    TermVote,
84    /// 保存当前 commit 索引(若有)。
85    CommitIndex,
86    /// 快照元数据:`(last_included_index, last_included_term)`。
87    SnapshotMeta,
88    /// 状态机快照原始字节。
89    SnapshotData,
90}
91
92// 键空间编码:决定条目有序扫描与元数据分区
93impl Key {
94    /// 编码为有序字节键。
95    pub fn encode(&self) -> Vec<u8> {
96        // 按键种类生成固定前缀(及可选索引),决定存储字典序
97        match self {
98            // 日志条目键:0x00 前缀 + 大端索引,保证按索引有序扫描
99            Key::Entry(index) => {
100                // 预分配 1 字节前缀 + 8 字节 u64
101                let mut buf = Vec::with_capacity(1 + 8);
102                // 条目键前缀,保证全部 Entry 排在元数据键之前
103                buf.push(0x00);
104                // 大端写入索引,字典序即索引序
105                buf.extend_from_slice(&index.to_be_bytes());
106                // 返回完整有序键
107                buf
108            }
109            // 任期/投票元数据键
110            Key::TermVote => vec![0x01],
111            // 已提交索引元数据键
112            Key::CommitIndex => vec![0x02],
113            // 快照元数据键(last_included_index/term)
114            Key::SnapshotMeta => vec![0x03],
115            // 快照数据键(状态机字节)
116            Key::SnapshotData => vec![0x04],
117        }
118    }
119}
120
121/// Raft 日志保存一系列任意命令(通常是写操作),在节点间复制,并顺序应用到本地状态机。
122/// 每条日志含索引、命令,以及领导者提出它时的任期。命令可为 noop(None),
123/// 在选出领导者时追加(见论文 5.4.2 节)。示例:
124///
125/// Index | Term | Command
126/// ------|------|------------------------------------------------------
127///   1   |   1  | None
128///   2   |   1  | CREATE TABLE table (id INT PRIMARY KEY, value STRING)
129///   3   |   1  | INSERT INTO table VALUES (1, 'foo')
130///   4   |   2  | None
131///   5   |   2  | UPDATE table SET value = 'bar' WHERE id = 1
132///   6   |   2  | DELETE FROM table WHERE id = 1
133///
134/// 注意这只是示意;实际命令不必是 SQL,而是任意底层写操作。
135///
136/// 使用键值存储按索引落盘日志条目,以及若干元数据键(例如本任期投票给了谁)。
137///
138/// 稳态下日志只追加:客户端提交命令后,领导者经 [`Log::append`] 写入本地日志,
139/// 再复制给跟随者,跟随者经 [`Log::splice`] 追加。当某索引已复制到多数节点时
140/// 即变为已提交,此前日志不可变,并保证最终所有节点都会拥有它。
141/// 节点通过 [`Log::commit`] 跟踪 commit 索引,并将已提交命令应用到状态机。
142///
143/// 但未提交条目可被替换或删除。领导者可能已追加却无法达成共识
144///(例如无法与多数节点通信)。若另选新领导者并在相同索引写入不同命令,
145/// 旧领导者或跟随者发现后,会用新领导者的条目替换未提交部分。
146///
147/// Raft 日志不变量:
148///
149/// * 条目索引从 1 起连续(无空洞)。
150/// * 条目任期相对前一条从不下降。
151/// * 条目任期不超过当前任期。
152/// * 追加的条目是持久的(刷盘)。
153/// * 追加的条目使用当前任期。
154/// * 已提交条目在快照后可截断前缀(见 [`Log::compact_to`])。
155/// * 已提交条目最终会复制到所有节点。
156/// * 相同索引/任期的条目含相同命令。
157/// * 若两份日志在某索引/任期匹配,则此前所有条目相同(见论文 5.3 节)。
158pub struct Log {
159    /// 底层存储引擎。使用 trait 对象而非泛型,以便运行时选择引擎,
160    /// 并避免把泛型参数传遍整个 Raft。
161    pub engine: Box<dyn storage::Engine>,
162    /// 当前任期。
163    term: Term,
164    /// 本任期的领导者投票(若有)。
165    vote: Option<NodeID>,
166    /// 日志中仍保留的第一条条目索引(截断后 > 1)。
167    first_index: Index,
168    /// 快照最后包含的任期(first_index-1 对应的 term;无快照时为 0)。
169    snapshot_term: Term,
170    /// 最后一条已存条目的索引。
171    last_index: Index,
172    /// 最后一条已存条目的任期。
173    last_term: Term,
174    /// 最后一条已提交条目的索引。
175    commit_index: Index,
176    /// 最后一条已提交条目的任期。
177    commit_term: Term,
178    /// 为 true 时,追加后 fsync 到磁盘。这是 Raft 要求的,但有明显性能代价
179    ///(尤其是未做批量 fsync 优化时)。关闭可大幅提升写性能,但崩溃可能丢数据,
180    /// 某些场景下会导致日志“未提交”与状态机分叉。
181    fsync: bool,
182}
183
184// 日志核心 API:追加/提交/拼接/压缩/快照与查询
185impl Log {
186    /// 使用给定存储引擎初始化日志。
187    pub fn new(mut engine: Box<dyn storage::Engine>) -> Result<Self> {
188        // 从磁盘加载初始内存状态。
189        // 读取持久化的 (term, vote);不存在则视为 term=0、未投票
190        let (term, vote) = engine
191            // 选举安全元数据键
192            .get(&Key::TermVote.encode())?
193            // Option<Vec<u8>> → Option<&[u8]>,便于借用解码
194            .as_deref()
195            // 有值则 bincode 还原 (Term, Option<NodeID>)
196            .map(decode_value)
197            // 把 Option<Result<_>> 展平为 Result<Option<_>>
198            .transpose()?
199            // 冷启动:无任期、未投票
200            .unwrap_or((0, None));
201        // 扫描全部条目,取最后一条作为 last_index/last_term;空日志则为 (0,0)
202        let (mut last_index, mut last_term) = engine
203            // 全量条目区间扫描(按 Entry 键字典序)
204            .scan_dyn((
205                // 条目键下界:索引 0 起(实际条目从 1 开始)
206                Bound::Included(Key::Entry(0).encode()),
207                // 条目键上界:最大 u64 索引
208                Bound::Included(Key::Entry(u64::MAX).encode()),
209            // 结束多行表达式
210            ))
211            // 取序最大的一条,即当前日志尾
212            .last()
213            // 存储错误向上冒泡
214            .transpose()?
215            // 解码尾条目
216            .map(|(_, v)| Entry::decode(&v))
217            // 解码错误向上冒泡
218            .transpose()?
219            // 抽出 (index, term) 作为 last 指针
220            .map(|e| (e.index, e.term))
221            // 空日志:尚未有任何条目
222            .unwrap_or((0, 0));
223        // 读取持久化的 commit 索引与任期;缺失则 (0,0)
224        let (mut commit_index, mut commit_term) = engine
225            // commit 元数据键
226            .get(&Key::CommitIndex.encode())?
227            // 借用字节切片
228            .as_deref()
229            // 解码 (commit_index, commit_term)
230            .map(decode_value)
231            // 展平 Result
232            .transpose()?
233            // 无 commit 记录:尚未提交任何条目
234            .unwrap_or((0, 0));
235        // 读取快照元数据;无快照时 (0,0)
236        let (snap_index, snapshot_term) = engine
237            // 快照基座元数据键
238            .get(&Key::SnapshotMeta.encode())?
239            // 借用字节切片
240            .as_deref()
241            // 解码 (last_included_index, last_included_term)
242            .map(decode_value)
243            // 展平 Result
244            .transpose()?
245            // 无快照:基座为 0
246            .unwrap_or((0, 0));
247        // first_index = 快照之后下一条;无快照且无日志时为 1。
248        let mut first_index = if snap_index > 0 {
249            // 有快照:保留日志从 last_included_index+1 开始
250            snap_index + 1
251        // 无快照且磁盘上也没有条目
252        } else if last_index == 0 {
253            // 无快照也无条目:约定从索引 1 起写
254            1
255        // 无快照但已有条目:从最小索引恢复 first
256        } else {
257            // 扫描最小 entry 索引
258            engine
259                // 同样扫全部 Entry 键区间
260                .scan_dyn((
261                    // 下界:最小可能条目键
262                    Bound::Included(Key::Entry(0).encode()),
263                    // 上界:最大可能条目键
264                    Bound::Included(Key::Entry(u64::MAX).encode()),
265                // 结束多行表达式
266                ))
267                // 取序最小的一条,即当前日志头
268                .next()
269                // 存储错误向上冒泡
270                .transpose()?
271                // 解码头条目
272                .map(|(_, v)| Entry::decode(&v))
273                // 解码错误向上冒泡
274                .transpose()?
275                // 头条目索引作为 first_index
276                .map(|e| e.index)
277                // 理论上不应为空,兜底为 1
278                .unwrap_or(1)
279        };
280
281        // 快照之后若无剩余日志条目,last/commit 至少要覆盖快照基座。
282        if snap_index > 0 {
283            // 快照已包含到 snap_index,本地 last 不能落后于基座
284            if last_index < snap_index {
285                // 抬升 last 到快照覆盖的最后索引
286                last_index = snap_index;
287                // last 任期与快照基座任期对齐
288                last_term = snapshot_term;
289            }
290            // 快照内容视为已提交,commit 至少推进到基座
291            if commit_index < snap_index {
292                // 抬升 commit 到快照覆盖点
293                commit_index = snap_index;
294                // commit 任期与快照基座任期对齐
295                commit_term = snapshot_term;
296            }
297            // 校正 first_index,保证与快照元数据一致
298            if first_index != snap_index + 1 {
299                // 强制 first 紧挨快照之后
300                first_index = snap_index + 1;
301            }
302        }
303
304        let fsync = true; // 默认开启 fsync
305        // 组装内存中的 Log 视图,与磁盘状态对齐
306        Ok(Self {
307            // 接管调用方传入的存储引擎
308            engine,
309            // 已恢复的当前任期
310            term,
311            // 已恢复的本任期投票
312            vote,
313            // 当前仍保留的日志起点
314            first_index,
315            // 快照基座任期(无快照时为 0)
316            snapshot_term,
317            // 日志尾索引
318            last_index,
319            // 日志尾任期
320            last_term,
321            // 已提交索引
322            commit_index,
323            // 已提交任期
324            commit_term,
325            // 是否在写后 fsync
326            fsync,
327        // 结束闭包/结构体表达式
328        })
329    }
330
331    /// 日志中仍保留的第一条索引。
332    pub fn get_first_index(&self) -> Index {
333        // 供复制/快照逻辑判断本地前缀是否已被压缩
334        self.first_index
335    }
336
337    /// 快照基座:`(last_included_index, last_included_term)`;无快照时 `(0,0)`。
338    pub fn get_snapshot_meta(&self) -> (Index, Term) {
339        // first_index<=1 表示尚未做压缩,无有效快照基座
340        if self.first_index <= 1 {
341            // 无快照:约定返回 (0,0)
342            (0, 0)
343        // 已压缩:暴露 last_included 供 InstallSnapshot/复制对齐
344        } else {
345            // 基座索引为 first_index-1,任期来自快照元数据
346            (self.first_index - 1, self.snapshot_term)
347        }
348    }
349
350    /// 截断并删除 `<= last_included_index` 的日志条目(快照后压缩)。
351    pub fn compact_to(&mut self, last_included_index: Index, last_included_term: Term) -> Result<()> {
352        // 安全约束:只能压缩已提交前缀,防止丢未提交数据
353        assert!(last_included_index <= self.commit_index, "compact beyond commit");
354        // 已压缩过或目标不推进:幂等返回
355        if last_included_index + 1 <= self.first_index && last_included_index > 0 {
356            // 目标基座不新于现有 first,无需再删
357            return Ok(());
358        }
359        // 逐条删除将被快照覆盖的前缀条目
360        for i in self.first_index..=last_included_index {
361            // 删除单条 Entry 键,释放已快照历史
362            self.engine.delete(&Key::Entry(i).encode())?;
363        }
364        // 持久化新的快照元数据
365        self.engine.set(
366            // 快照基座键
367            &Key::SnapshotMeta.encode(),
368            // 编码 (last_included_index, last_included_term)
369            encode_value(&(last_included_index, last_included_term)),
370        // 元数据写入失败则压缩中止
371        )?;
372        // 按配置刷盘,保证压缩结果崩溃可恢复
373        if self.fsync {
374            // 强制落盘,避免崩溃后仍见已删前缀
375            self.engine.flush()?;
376        }
377        // 内存 first_index 推进到快照之后
378        self.first_index = last_included_index + 1;
379        // 记录基座任期,供 has()/InstallSnapshot 匹配
380        self.snapshot_term = last_included_term;
381        // 若本地日志本就短于快照,同步 last 指针到基座
382        if self.last_index < last_included_index {
383            // last 至少等于快照覆盖点
384            self.last_index = last_included_index;
385            // last 任期同步为基座任期
386            self.last_term = last_included_term;
387        }
388        // 压缩完成
389        Ok(())
390    }
391
392    /// 安装快照后重置日志:丢弃全部条目,仅保留快照基座。
393    pub fn reset_with_snapshot(
394        // 可变借用自身以改内存视图与引擎
395        &mut self,
396        // 快照覆盖到的最后日志索引
397        last_included_index: Index,
398        // 该索引对应条目的任期
399        last_included_term: Term,
400    // 安装失败时调用方应中止应用快照
401    ) -> Result<()> {
402        // 删除所有 entry
403        // 先收集全部条目键,避免边扫边删
404        let to_delete: Vec<_> = self
405            // 访问底层引擎
406            .engine
407            // 扫描全部 Entry 键
408            .scan_dyn((
409                // 条目下界
410                Bound::Included(Key::Entry(0).encode()),
411                // 条目上界
412                Bound::Included(Key::Entry(u64::MAX).encode()),
413            // 结束多行表达式
414            ))
415            // 只保留键;忽略单条扫描错误以免中断清空
416            .filter_map(|r| r.ok().map(|(k, _)| k))
417            // 物化键列表后再删
418            .collect();
419        // 清空本地条目:快照已覆盖历史
420        for k in to_delete {
421            // 删除收集到的每一条 Entry
422            self.engine.delete(&k)?;
423        }
424        // 写入快照元数据
425        self.engine.set(
426            // 快照基座键
427            &Key::SnapshotMeta.encode(),
428            // 编码新基座 (index, term)
429            encode_value(&(last_included_index, last_included_term)),
430        // 基座元数据必须先落库
431        )?;
432        // 快照内容视为已提交
433        self.commit_index = last_included_index;
434        // commit 任期对齐快照基座
435        self.commit_term = last_included_term;
436        // 持久化 commit,重启后与状态机对齐
437        self.engine.set(
438            // commit 元数据键
439            &Key::CommitIndex.encode(),
440            // 编码当前 (commit_index, commit_term)
441            encode_value(&(self.commit_index, self.commit_term)),
442        // commit 与快照基座一并持久,避免重启分叉
443        )?;
444        // 按配置刷盘,保证 InstallSnapshot 结果持久
445        if self.fsync {
446            // 强制落盘
447            self.engine.flush()?;
448        }
449        // 内存视图重置为「仅有快照基座」
450        self.first_index = last_included_index + 1;
451        // 基座任期
452        self.snapshot_term = last_included_term;
453        // last 停在快照覆盖点(其后尚无条目)
454        self.last_index = last_included_index;
455        // last 任期与基座一致
456        self.last_term = last_included_term;
457        // 重置完成
458        Ok(())
459    }
460
461    /// 控制是否对写入做 fsync。关闭可能违反 Raft 保证,见 fsync 字段注释。
462    pub fn enable_fsync(&mut self, fsync: bool) {
463        // 运行时开关:测试可关,生产应开
464        self.fsync = fsync
465    }
466
467    /// 返回 commit 索引与任期。
468    pub fn get_commit_index(&self) -> (Index, Term) {
469        // 供节点状态机 apply 与心跳携带 leaderCommit
470        (self.commit_index, self.commit_term)
471    }
472
473    /// 返回最后一条日志的索引与任期。
474    pub fn get_last_index(&self) -> (Index, Term) {
475        // 供选举 RequestVote 与复制进度比较
476        (self.last_index, self.last_term)
477    }
478
479    /// 返回当前任期(无则为 0)与投票。
480    pub fn get_term_vote(&self) -> (Term, Option<NodeID>) {
481        // 供消息 term 校验与投票决策
482        (self.term, self.vote)
483    }
484
485    /// 保存当前任期与投票(若有)。强制任期不回退,且一个任期内只投一票。
486    /// append() 使用此任期;splice() 不能写入超过该任期的条目。
487    pub fn set_term_vote(&mut self, term: Term, vote: Option<NodeID>) -> Result<()> {
488        // term 0 非法:协议从 1 起
489        assert!(term > 0, "can't set term 0");
490        // 任期只能单调不减,防止时钟回拨式脑裂
491        assert!(term >= self.term, "term regression {} → {}", self.term, term);
492        // 同一任期内已投票则不可改投他人
493        assert!(term > self.term || self.vote.is_none() || vote == self.vote, "can't change vote");
494
495        // 幂等:无变化则跳过写盘
496        if term == self.term && vote == self.vote {
497            // 已是目标状态,避免无意义 fsync
498            return Ok(());
499        }
500        // 持久化 (term, vote),选举安全的关键
501        self.engine.set(&Key::TermVote.encode(), encode_value(&(term, vote)))?;
502        // 即使 Log::fsync = false 也总是 fsync。任期变更很少,对性能影响不大,
503        // 而双重投票可能导致多领导者与脑裂,后果严重。
504        self.engine.flush()?;
505        // 更新内存视图
506        self.term = term;
507        // 记录本任期投票对象(可为 None 表示仅升任期)
508        self.vote = vote;
509        // 任期/投票已持久
510        Ok(())
511    }
512
513    /// 在当前任期向日志追加命令并刷盘,返回其索引。
514    /// None 表示 noop 命令,通常在 Raft 领导者变更后使用。
515    pub fn append(&mut self, command: Option<Vec<u8>>) -> Result<Index> {
516        // 普通客户端/noop 写:无成员变更字段
517        self.append_entry(command, None)
518    }
519
520    /// 追加一条成员配置变更日志。
521    pub fn append_membership(&mut self, membership: MembershipEntry) -> Result<Index> {
522        // 成员变更条目:command 为空,仅携带 membership
523        self.append_entry(None, Some(membership))
524    }
525
526    /// 追加完整条目字段。
527    pub fn append_entry(
528        // 可变借用以推进 last 并写引擎
529        &mut self,
530        // 状态机命令;noop 或成员变更时为 None
531        command: Option<Vec<u8>>,
532        // 成员配置;普通命令时为 None
533        membership: Option<MembershipEntry>,
534    // 返回新条目索引;写盘失败则不推进 last
535    ) -> Result<Index> {
536        // 领导者必须已进入有效任期才能提出条目
537        assert!(self.term > 0, "can't append entry in term 0");
538        // 业务命令与成员变更互斥,避免一条日志语义歧义
539        assert!(
540            // 至多一种载荷
541            command.is_none() || membership.is_none(),
542            // 违反互斥则条目语义无法解释
543            "command and membership are mutually exclusive"
544        );
545        // 在 last 之后连续追加,任期取当前领导者任期
546        let entry = Entry {
547            // 新索引紧接 last,保证连续性
548            index: self.last_index + 1,
549            // 领导者当前任期
550            term: self.term,
551            // 客户端命令或 noop
552            command,
553            // 可选成员变更
554            membership,
555        };
556        // 按索引键写入引擎
557        self.engine.set(&Key::Entry(entry.index).encode(), entry.encode())?;
558        // Raft 要求追加持久化后再复制/应答
559        if self.fsync {
560            // 追加必须落盘,崩溃后不能丢未复制承诺
561            self.engine.flush()?;
562        }
563        // 推进 last 指针,供后续 append/心跳使用
564        self.last_index = entry.index;
565        // 同步 last 任期
566        self.last_term = entry.term;
567        // 返回新条目索引给上层
568        Ok(entry.index)
569    }
570
571    /// 从日志中扫描最新的成员配置条目(若有)。
572    pub fn latest_membership(&mut self) -> Result<Option<(Index, MembershipEntry)>> {
573        // 线性扫描找最后一条 membership(配置量少,可接受)
574        let mut found = None;
575        // 从 1 扫到 last,覆盖全量保留日志
576        for entry in self.scan(1..=self.last_index) {
577            // 单条解码/存储错误向上返回
578            let entry = entry?;
579            // 后出现的配置覆盖先前结果
580            if let Some(m) = entry.membership {
581                // 记录最新 (index, membership)
582                found = Some((entry.index, m));
583            }
584        }
585        // 无成员变更条目时为 None
586        Ok(found)
587    }
588
589    /// 提交到给定索引(含)。该索引必须存在且不早于当前 commit 索引。
590    pub fn commit(&mut self, index: Index) -> Result<Index> {
591        // 取出目标条目任期,并校验不回退、条目存在
592        let term = match self.get(index)? {
593            // 禁止 commit 索引回退
594            Some(entry) if entry.index < self.commit_index => {
595                // 违反单调性:协议层 bug
596                panic!("commit index regression {} → {}", self.commit_index, entry.index);
597            }
598            // 已提交到该点:幂等
599            Some(entry) if entry.index == self.commit_index => return Ok(index),
600            // 正常推进:记录该条目任期
601            Some(entry) => entry.term,
602            // 提交不存在的索引是严重错误
603            None => panic!("commit index {index} does not exist"),
604        };
605        // 持久化 (commit_index, commit_term)
606        self.engine.set(&Key::CommitIndex.encode(), encode_value(&(index, term)))?;
607        // 注意:commit 索引不必 fsync,因为条目已 fsync,且可从多数派日志恢复。
608        // 更新内存 commit 视图,驱动状态机 apply
609        self.commit_index = index;
610        // 同步 commit 任期,供快照/状态查询
611        self.commit_term = term;
612        // 返回新的 commit 索引
613        Ok(index)
614    }
615
616    /// 获取指定索引的条目;不存在则返回 None。
617    pub fn get(&mut self, index: Index) -> Result<Option<Entry>> {
618        // 按 Entry 键读取并解码;缺失返回 None
619        self.engine.get(&Key::Entry(index).encode())?.map(|v| Entry::decode(&v)).transpose()
620    }
621
622    /// 检查日志是否包含给定索引与任期的条目。
623    pub fn has(&mut self, index: Index, term: Term) -> Result<bool> {
624        // 快路径:与 last_index 比较。跟随者处理 append/心跳时的常见情况。
625        // 索引 0 或超过本地 last:肯定不存在
626        if index == 0 || index > self.last_index {
627            // prevLogIndex 无效或本地更短
628            return Ok(false);
629        }
630        // 快照基座
631        // 恰好落在 last_included 且 term 匹配:视为存在(条目已压缩)
632        if index + 1 == self.first_index && term == self.snapshot_term && index > 0 {
633            // 压缩前缀上的匹配成功,用于 prevLog 校验
634            return Ok(true);
635        }
636        // 已压缩掉且不是基座:无法匹配
637        if index < self.first_index {
638            // 历史已删且 term 对不上基座
639            return Ok(false);
640        }
641        // 与 last 完全一致的快路径
642        if (index, term) == (self.last_index, self.last_term) {
643            // 常见:心跳 prev 正好是本地尾
644            return Ok(true);
645        }
646        // 回落到盘读取该索引并比对任期
647        Ok(self.get(index)?.map(|e| e.term == term).unwrap_or(false))
648    }
649
650    /// 返回给定索引范围内的日志条目迭代器。
651    pub fn scan(&mut self, range: impl RangeBounds<Index>) -> Iterator<'_> {
652        // 规范化边界,避免 BTreeMap range start > end panic。
653        // 将 RangeBounds 转为闭区间 [start_idx, end_idx_inclusive]
654        let start_idx = match range.start_bound() {
655            // 半开下界:下一条起
656            Bound::Excluded(&i) => i.saturating_add(1),
657            // 闭下界:含该索引
658            Bound::Included(&i) => i,
659            // 无下界:从 0 起(实际条目从 1)
660            Bound::Unbounded => 0,
661        };
662        // 规范化上界为闭区间终点
663        let end_idx_inclusive = match range.end_bound() {
664            // 半开上界:前一条为止
665            Bound::Excluded(&i) => i.saturating_sub(1),
666            // 闭上界:含该索引
667            Bound::Included(&i) => i,
668            // 无上界:扫到最大索引
669            Bound::Unbounded => Index::MAX,
670        };
671        // 空区间:返回空迭代器
672        if start_idx > end_idx_inclusive {
673            // 调用方 range 非法或空,避免引擎 panic
674            return Iterator::new(Box::new(std::iter::empty()));
675        }
676        // 映射为存储层有序键区间
677        let from = Bound::Included(Key::Entry(start_idx).encode());
678        // 上界同样编码为 Entry 键
679        let to = Bound::Included(Key::Entry(end_idx_inclusive).encode());
680        // 包装为 Entry 解码迭代器
681        Iterator::new(self.engine.scan_dyn((from, to)))
682    }
683
684    /// 返回可应用条目的迭代器:从当前 applied 索引之后到 commit 索引。
685    pub fn scan_apply(&mut self, applied_index: Index) -> Iterator<'_> {
686        // 注意:不断言 commit_index >= applied_index,因为本地 commit 索引不刷盘——
687        // 重启丢失后可从多数派日志恢复。
688        // 状态机已追平 commit:无可应用条目
689        if applied_index >= self.commit_index {
690            // 空迭代,调用方无需 apply
691            return Iterator::new(Box::new(std::iter::empty()));
692        }
693        // 扫描 (applied, commit] 区间,顺序应用到状态机
694        self.scan(applied_index + 1..=self.commit_index)
695    }
696
697    /// 将一组条目拼接到日志并刷盘。新索引会追加。
698    /// 重叠且任期相同的索引必须相等并被忽略;重叠但任期不同时,
699    /// 在首个冲突处截断现有日志,再拼接新条目。
700    ///
701    /// 条目索引必须连续、任期相等或递增;首条索引须在 [1, last_index+1] 内,
702    /// 任期不低于前一条(base)且不超过当前任期。
703    pub fn splice(&mut self, entries: Vec<Entry>) -> Result<Index> {
704        // 空输入不改变日志
705        let (Some(first), Some(last)) = (entries.first(), entries.last()) else {
706            return Ok(self.last_index); // 空输入为 no-op
707        };
708
709        // 检查条目形态是否合法。
710        // 索引/任期从 1 起,0 非法
711        assert!(first.index > 0 && first.term > 0, "spliced entry has index or term 0",);
712        // 索引必须严格连续,无空洞
713        assert!(
714            // 相邻条目 index 差必须为 1
715            entries.windows(2).all(|w| w[0].index + 1 == w[1].index),
716            // 空洞会破坏日志匹配与 commit 推进
717            "spliced entries are not contiguous"
718        );
719        // 批内任期单调不减
720        assert!(
721            // 不允许后一条 term 小于前一条
722            entries.windows(2).all(|w| w[0].term <= w[1].term),
723            // 批内 term 回退违反 Raft 日志属性
724            "spliced entries have term regression",
725        );
726
727        // 检查条目能否接到现有日志,且任期不回退。
728        // 不能写入超过本节点已知当前任期的条目
729        assert!(last.term <= self.term, "splice term {} beyond current {}", last.term, self.term);
730        // 与 base 条目衔接:任期不回退,且必须贴住现有日志或从 1 起
731        match self.get(first.index - 1)? {
732            // 前一条任期更高:违反 term 不降
733            Some(base) if first.term < base.term => {
734                // 协议层错误:领导者不应下发回退任期
735                panic!("splice term regression {} → {}", base.term, first.term)
736            }
737            // base 存在且任期合法:可拼接
738            Some(_) => {}
739            // 从日志起点开始追加
740            None if first.index == 1 => {}
741            // 中间空洞:违反日志连续性
742            None => panic!("first index {} must touch existing log", first.index),
743        }
744
745        // 跳过日志中已存在的条目。
746        // 剩余待写入切片(跳过与本地一致的前缀)
747        let mut entries = entries.as_slice();
748        // 扫描重叠区间,比对 index/term/command
749        let mut scan = self.scan(first.index..=last.index);
750        // 逐条与本地重叠前缀比对
751        while let Some(entry) = scan.next().transpose()? {
752            // [0] 合法,因为扫描范围与 entries 大小相同。
753            // 索引应对齐
754            assert!(entry.index == entries[0].index, "index mismatch at {entry:?}");
755            // 任期冲突:自此截断并重写
756            if entry.term != entries[0].term {
757                // 停止跳过,entries 余下部分将覆盖冲突尾
758                break;
759            }
760            // 同 index/term 则命令与成员字段必须一致(日志匹配属性)
761            assert!(
762                // 相同 (index,term) 必须同 command/membership
763                entry.command == entries[0].command && entry.membership == entries[0].membership,
764                // 违反日志匹配属性:同 index/term 内容必须唯一
765                "command/membership mismatch at {entry:?}"
766            );
767            // 跳过已匹配条目
768            entries = &entries[1..];
769        }
770        // 释放扫描对 engine 的借用,后续才能写
771        drop(scan);
772
773        // 若全部已存在则完成。
774        let Some(first) = entries.first() else {
775            // 重叠前缀完全一致且无新条目,last 不变
776            return Ok(self.last_index);
777        };
778
779        // 写入尚未存在的条目,并删除旧日志尾部(若有)。
780        // 不能写到 commit 索引以下,那些条目必须不可变。
781        assert!(first.index > self.commit_index, "spliced entries below commit index");
782
783        // 写入冲突点及之后的新条目
784        for entry in entries {
785            // 覆盖或追加 Entry 键
786            self.engine.set(&Key::Entry(entry.index).encode(), entry.encode())?;
787        }
788        // 删除新 last 之后的旧尾部(未提交分歧日志)
789        for index in last.index + 1..=self.last_index {
790            // 截断本地更长的冲突后缀
791            self.engine.delete(&Key::Entry(index).encode())?;
792        }
793        // 刷盘保证一致性复制结果持久
794        if self.fsync {
795            // 冲突解决结果必须落盘
796            self.engine.flush()?;
797        }
798
799        // 更新 last 指针到拼接结果末尾
800        self.last_index = last.index;
801        // 同步 last 任期为批末条目任期
802        self.last_term = last.term;
803        // 返回新的 last_index
804        Ok(self.last_index)
805    }
806
807    /// 返回日志引擎状态。
808    pub fn status(&mut self) -> Result<storage::Status> {
809        // 透传底层存储状态(供节点 Status 响应)
810        self.engine.status()
811    }
812}
813
814/// 日志条目迭代器。
815pub struct Iterator<'a> {
816    // 底层存储扫描迭代器,按键序产出 (key, value)
817    inner: Box<dyn storage::ScanIterator + 'a>,
818}
819
820// 构造期:仅包装底层 ScanIterator
821impl<'a> Iterator<'a> {
822    // 包装存储扫描为 Entry 迭代器
823    fn new(inner: Box<dyn storage::ScanIterator + 'a>) -> Self {
824        // 持有动态扫描器,生命周期绑定到 Log 引擎借用
825        Self { inner }
826    }
827}
828
829// 标准迭代协议:按索引序产出已解码条目
830impl std::iter::Iterator for Iterator<'_> {
831    // 每次产出解码后的条目或存储/解码错误
832    type Item = Result<Entry>;
833
834    // 推进底层扫描并解码为 Entry
835    fn next(&mut self) -> Option<Self::Item> {
836        // 忽略键,只解码 value 为 Entry
837        self.inner.next().map(|r| r.and_then(|(_, v)| Entry::decode(&v)))
838    }
839}