raft-rust 0.1.0

Standalone Raft consensus library in Rust
Documentation
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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
// 内存 keydir:有序映射键 → 值位置
use std::collections::BTreeMap;
// keydir 范围扫描的迭代器类型
use std::collections::btree_map::Range;
// 底层日志文件句柄
use std::fs::File;
// 缓冲读写与定位
use std::io::{BufReader, BufWriter, Read as _, Seek as _, SeekFrom, Write as _};
// 范围边界,用于 scan
use std::ops::{Bound, RangeBounds};
// 日志文件路径
use std::path::PathBuf;
// 构建 keydir 时用 std IO Result,与库 Result 区分
use std::result::Result as StdResult;

// 跨进程排他锁,防止双开同一数据文件
use fs4::fs_std::FileExt;
// 打开/压缩/截断时的运维日志
use log::{error, info};

// Engine trait 与对外 Status
use super::{Engine, Status};
// 库错误类型
use crate::error::{Error, Result};

/// BitCask 的极简变体。BitCask 是日志结构键值引擎(如 Riak 所用)。
/// 与其他实现生成的 BitCask 库不兼容。参见:
/// <https://riak.com/assets/bitcask-intro.pdf>
///
/// BitCask 将键值对追加写入日志文件,并在内存中维护键到文件偏移的映射。
/// 所有存活键须能装入内存。删除会向日志写入墓碑值。为清理旧垃圾
///(已删除或被覆盖的键),可通过只写入存活数据的新日志来压缩,丢弃被替换的值与墓碑。
///
/// 本实现比标准 BitCask 简单得多:
///
/// * 不写多个固定大小日志文件,而是使用单个任意大小的追加日志。
///   这会增加压缩量(每次压缩须重写整文件),也可能超过文件系统文件大小限制。
///   不过本库预期数据库较小。
///
/// * 压缩期间会锁住数据库的读写。可接受:仅在节点启动时压缩,且文件预期较小。
///
/// * 不使用 hint 文件,打开时扫描日志本身构建 keydir。Hint 只省略值,
///   而本库值预期较小,hint 几乎与压缩后的日志一样大。
///
/// * 日志条目不含时间戳或校验和。
///
/// 编码后的日志条目结构:
///
/// 1. 键长度,大端 u32 [4 字节]。
/// 2. 值长度,大端 i32;墓碑为 -1 [4 字节]。
/// 3. 键原始字节 [<= 2 GB]。
/// 4. 值原始字节 [<= 2 GB]。
pub struct BitCask {
    /// 当前追加写日志文件。
    log: Log,
    /// 将键映射到 [`BitCask::log`] 中值的偏移与长度。
    keydir: KeyDir,
// 结束当前作用域
}

/// 将键映射到日志文件中值的位置。
// 使用 BTreeMap 以支持有序范围扫描(Raft 日志按键序迭代)
type KeyDir = BTreeMap<Vec<u8>, ValueLocation>;

/// 日志文件中值的位置。
// Copy 便于从 keydir 取出后按位置随机读
#[derive(Clone, Copy)]
// 定义数据结构
struct ValueLocation {
    /// 值在日志文件中的字节偏移。
    offset: u64,
    /// 值的字节长度。
    length: usize,
// 结束当前作用域
}

// 为类型实现方法
impl ValueLocation {
    // 值在文件中的结束偏移(不含),用于 EOF 校验
    fn end(&self) -> u64 {
        // offset + length
        self.offset + self.length as u64
    // 结束当前作用域
    }
// 结束当前作用域
}

// 为类型实现方法
impl BitCask {
    /// 在给定路径打开或创建 BitCask 数据库。
    pub fn new(path: PathBuf) -> Result<Self> {
        // 打开/创建追加日志
        let mut log = Log::new(path.clone())?;
        // 扫描日志重建内存索引
        let keydir = log.build_keydir()?;
        // 记录存活键数量,便于启动观测
        info!("Opened {} with {} live keys", path.display(), keydir.len());
        // 组装引擎实例
        Ok(Self { log, keydir })
    // 结束当前作用域
    }

    /// 打开 BitCask 数据库;若打开时垃圾占比与字节数超过阈值则自动压缩。
    pub fn new_maybe_compact(
        // 列表或字段续项
        path: PathBuf,
        // 垃圾占比阈值(0.0–1.0)
        garbage_min_fraction: f64,
        // 垃圾绝对字节阈值
        garbage_min_bytes: u64,
    // 业务逻辑步骤
    ) -> Result<Self> {
        // 先正常打开
        let mut engine = Self::new(path)?;

        // 读取盘占用与存活统计
        let status = engine.status()?;
        // 磁盘总大小(含垃圾)
        let total_size = status.disk_size;
        // 垃圾字节数
        let garbage_size = status.garbage_disk_size();
        // 垃圾占比
        let garbage_fraction = garbage_size as f64 / total_size as f64;
        // 同时满足:有垃圾、字节数与占比都超阈值才压缩
        if garbage_size > 0
            // 业务逻辑步骤
            && garbage_size >= garbage_min_bytes
            // 业务逻辑步骤
            && garbage_fraction >= garbage_min_fraction
        // 进入代码块
        {
            // 压缩前记录预期可回收空间
            info!(
                // 列表或字段续项
                "Compacting {} to remove {:.0}% garbage ({:.1} MB out of {:.1} MB)",
                // 列表或字段续项
                engine.log.path.display(),
                // 列表或字段续项
                garbage_fraction * 100.0,
                // 列表或字段续项
                garbage_size as f64 / 1024.0 / 1024.0,
                // 业务逻辑步骤
                total_size as f64 / 1024.0 / 1024.0
            // 结束调用参数列表
            );
            // 重写仅含存活键的新日志
            engine.compact()?;
            // 压缩后记录目标大小
            info!(
                // 列表或字段续项
                "Compacted {} to size {:.1} MB",
                // 列表或字段续项
                engine.log.path.display(),
                // 业务逻辑步骤
                (total_size - garbage_size) as f64 / 1024.0 / 1024.0
            // 结束调用参数列表
            );
        // 结束当前作用域
        }

        // 返回可能已压缩的引擎
        Ok(engine)
    // 结束当前作用域
    }
// 结束当前作用域
}

// 实现通用 Engine 接口,供 Raft Log 使用
impl Engine for BitCask {
    // 关联扫描迭代器
    type ScanIterator<'a> = ScanIterator<'a>;

    // 删除:写墓碑并从 keydir 移除
    fn delete(&mut self, key: &[u8]) -> Result<()> {
        // 追加墓碑条目(value = None)
        self.log.write_entry(key, None)?;
        // 内存索引删除,后续 get 视为不存在
        self.keydir.remove(key);
        // 成功返回
        Ok(())
    // 结束当前作用域
    }

    // 将日志刷到稳定存储
    fn flush(&mut self) -> Result<()> {
        // 测试中不做 fsync 以加速。在此禁用而非在测试里设
        // raft::Log::fsync = false,以便断言即使 flush 是空操作也会刷盘。
        #[cfg(not(test))]
        // 生产路径:fsync 整个文件
        self.log.file.sync_all()?;
        // 成功返回
        Ok(())
    // 结束当前作用域
    }

    // 按 keydir 定位后从日志随机读值
    fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        // 无索引项则键不存在
        let Some(location) = self.keydir.get(key) else {
            // 提前返回
            return Ok(None);
        // 结束当前作用域
        };
        // 按偏移读取值并包装为 Some
        self.log.read_value(*location).map(Some)
    // 结束当前作用域
    }

    // 在 keydir 上做有序范围扫描,值懒加载
    fn scan(&mut self, range: impl RangeBounds<Vec<u8>>) -> Self::ScanIterator<'_> {
        // 持有 keydir 范围迭代器 + 日志可变借用
        ScanIterator { inner: self.keydir.range(range), log: &mut self.log }
    // 结束当前作用域
    }

    // dyn Engine 路径:装箱静态 scan 结果
    fn scan_dyn(
        // 借用参数
        &mut self,
        // 列表或字段续项
        range: (Bound<Vec<u8>>, Bound<Vec<u8>>),
    // 业务逻辑步骤
    ) -> Box<dyn super::ScanIterator + '_> {

        // 复用 scan 实现
        Box::new(self.scan(range))
    // 结束当前作用域
    }

    // 设置/覆盖键:追加新值并更新 keydir
    fn set(&mut self, key: &[u8], value: Vec<u8>) -> Result<()> {
        // 写日志并拿到新值位置
        let value_location = self.log.write_entry(key, Some(&*value))?;
        // 覆盖内存索引(旧位置成为垃圾,待压缩回收)
        self.keydir.insert(key.to_vec(), value_location);
        // 成功返回
        Ok(())
    // 结束当前作用域
    }

    // 汇总逻辑大小与磁盘占用,供 Status 与压缩阈值
    fn status(&mut self) -> Result<Status> {
        // 存活键数
        let keys = self.keydir.len() as u64;
        // 逻辑大小:各存活键长 + 值长之和
        let size =
            // 链式变换结果
            self.keydir.iter().map(|(key, value_loc)| (key.len() + value_loc.length) as u64).sum();
        // 文件实际字节数(含垃圾)
        let disk_size = self.log.file.metadata()?.len();
        // 存活盘占用估算:逻辑大小 + 每键 8 字节长度前缀
        let live_disk_size = size + 8 * keys; // 计入长度前缀
        // 引擎名固定为 bitcask
        Ok(Status { name: "bitcask".to_string(), keys, size, disk_size, live_disk_size })
    // 结束当前作用域
    }
// 结束当前作用域
}

// 为类型实现方法
impl BitCask {
    /// 压缩当前日志:写出仅含存活键的新日志,并替换当前文件。
    pub fn compact(&mut self) -> Result<()> {
        // 创建临时日志文件;若已存在则截断。
        // 同目录 `.new` 扩展名,便于 rename 替换
        let new_path = self.log.path.with_extension("new");
        // 打开临时日志
        let mut new_log = Log::new(new_path)?;
        // 确保为空文件
        new_log.file.set_len(0)?;

        // 将全部存活条目写入新日志,并生成新 KeyDir。
        let mut new_keydir = KeyDir::new();
        // 遍历当前存活键
        for (key, value_loc) in &self.keydir {
            // 从旧日志读出当前值
            let value = self.log.read_value(*value_loc)?;
            // 写入新日志并记录新位置
            let value_loc = new_log.write_entry(key, Some(&value))?;
            // 填入新索引
            new_keydir.insert(key.clone(), value_loc);
        // 结束当前作用域
        }

        // 用新日志替换当前日志。
        // 原子 rename 到原路径
        std::fs::rename(&new_log.path, &self.log.path)?;
        // 更新临时 Log 的 path 字段为正式路径
        new_log.path = self.log.path.clone();

        // 切换到新日志与新 keydir
        self.log = new_log;
        // 业务逻辑步骤
        self.keydir = new_keydir;
        // 成功返回
        Ok(())
    // 结束当前作用域
    }
// 结束当前作用域
}

/// 数据库关闭时尝试刷盘。
impl Drop for BitCask {
    // 定义函数
    fn drop(&mut self) {
        // 尽力 fsync;失败只记日志,析构不能返回错误
        if let Err(error) = self.flush() {
            // 业务逻辑步骤
            error!("failed to flush file: {}", error)
        // 结束当前作用域
        }
    // 结束当前作用域
    }
// 结束当前作用域
}

/// BitCask 范围扫描迭代器。
pub struct ScanIterator<'a> {
    /// keydir 的范围迭代器。
    inner: Range<'a, Vec<u8>, ValueLocation>,
    /// 用于按位置读取值的日志引用。
    log: &'a mut Log,
// 结束当前作用域
}

// 为类型实现方法
impl ScanIterator<'_> {
    /// 真正从数据文件按位置读取值。
    fn map(&mut self, item: (&Vec<u8>, &ValueLocation)) -> <Self as Iterator>::Item {
        // 解构键与位置
        let (key, value_loc) = item;
        // 克隆键并从日志读值
        Ok((key.clone(), self.log.read_value(*value_loc)?))
    // 结束当前作用域
    }
// 结束当前作用域
}

// 正向迭代:按键升序产出
impl Iterator for ScanIterator<'_> {
    // 每项为可能失败的 (key, value)
    type Item = Result<(Vec<u8>, Vec<u8>)>;

    /// 迭代器逐步从数据文件读取数据。
    fn next(&mut self) -> Option<Self::Item> {
        // 取下一个 keydir 项并加载值
        self.inner.next().map(|item| self.map(item))
    // 结束当前作用域
    }
// 结束当前作用域
}

// 反向迭代:支持从尾部扫描 Raft 日志等场景
impl DoubleEndedIterator for ScanIterator<'_> {
    // 定义函数
    fn next_back(&mut self) -> Option<Self::Item> {
        // 从范围高端取项并加载值
        self.inner.next_back().map(|item| self.map(item))
    // 结束当前作用域
    }
// 结束当前作用域
}

/// BitCask 追加写日志文件,包含按如下格式编码的键值条目序列:
///
/// 1. 键长度,大端 u32 [4 字节]。
/// 2. 值长度,大端 i32;墓碑为 -1 [4 字节]。
/// 3. 键原始字节 [<= 2 GB]。
/// 4. 值原始字节 [<= 2 GB]。
struct Log {
    /// 已打开的日志文件。
    file: File,
    /// 日志文件路径。
    path: PathBuf,
// 结束当前作用域
}

// 为类型实现方法
impl Log {
    /// 打开日志文件;不存在则创建。在关闭前对文件加排他锁;
    /// 若锁已被持有则报错。
    fn new(path: PathBuf) -> Result<Self> {
        // 确保父目录存在
        if let Some(dir) = path.parent() {
            // 业务逻辑步骤
            std::fs::create_dir_all(dir)?
        // 结束当前作用域
        }
        // 读写创建、不截断,以便恢复已有日志
        let file = std::fs::OpenOptions::new()
            // 方法链调用
            .read(true)
            // 方法链调用
            .write(true)
            // 方法链调用
            .create(true)
            // 方法链调用
            .truncate(false)
            // 方法链调用
            .open(&path)?;
        // 排他锁:同一数据目录只允许一个进程打开
        if !file.try_lock_exclusive()? {
            // 提前返回
            return Err(Error::IO(format!("file {path:?} is already is use")));
        // 结束当前作用域
        }
        // 返回打开的日志
        Ok(Self { file, path })
    // 结束当前作用域
    }

    /// 扫描日志文件构建 keydir。若遇到不完整条目,视为不完整写入,截断文件余下部分。
    fn build_keydir(&mut self) -> Result<KeyDir> {
        // 复用 4 字节缓冲读长度字段
        let mut len_buf = [0u8; 4];
        // 累积最终内存索引
        let mut keydir = KeyDir::new();
        // 文件总长度,作为扫描上界
        let file_len = self.file.metadata()?.len();
        // 从文件偏移 0 开始,用 BufReader 逐条读取记录。
        let mut r = BufReader::new(&mut self.file);
        // 当前条目起始偏移
        let mut offset = r.seek(SeekFrom::Start(0))?;

        // 顺序扫描直到文件尾
        while offset < file_len {
            // 读取下一条目,返回键与值位置;墓碑返回 None。
            // 闭包内用 std IO 错误,便于识别 UnexpectedEof
            let result = || -> StdResult<(Vec<u8>, Option<ValueLocation>), std::io::Error> {
                // 读取键长度:4 字节 u32。
                r.read_exact(&mut len_buf)?;
                // 大端键长度
                let key_len = u32::from_be_bytes(len_buf);

                // 读取值长度:4 字节 i32,墓碑为 -1。
                r.read_exact(&mut len_buf)?;
                // 负长度 → 墓碑;否则记录值在文件中的位置
                let value_loc = match i32::from_be_bytes(len_buf) {
                    // 值为 -1(..0)表示删除标记,索引会从 keydir 移除该键。
                    ..0 => None, // 墓碑
                    // 值从 header(8) + key 之后开始
                    len => Some(ValueLocation {
                        // 列表或字段续项
                        offset: offset + 8 + key_len as u64,
                        // 列表或字段续项
                        length: len as usize,
                    // 列表或字段续项
                    }),
                // 结束当前作用域
                };

                // 读取键。
                let mut key: Vec<u8> = vec![0; key_len as usize];
                // 业务逻辑步骤
                r.read_exact(&mut key)?;

                // 跳过值。
                if let Some(value_loc) = value_loc {
                    // 值越过 EOF 视为截断写入
                    if value_loc.end() > file_len {
                        // 提前返回
                        return Err(std::io::Error::new(
                            // 列表或字段续项
                            std::io::ErrorKind::UnexpectedEof,
                            // 列表或字段续项
                            "value extends beyond end of file",
                        // 语法续行
                        ));
                    // 结束当前作用域
                    }
                    // 索引只需知道值位置,用 seek 跳过值内容以加速扫描。
                    r.seek_relative(value_loc.length as i64)?;
                // 结束当前作用域
                }

                // 更新文件偏移。
                // 下一条起点 = 本条起点 + 头 8 + 键 + 值(墓碑无值)
                offset += 8 + key_len as u64 + value_loc.map_or(0, |v| v.length) as u64;

                // 返回本条解析结果
                Ok((key, value_loc))
            // 业务逻辑步骤
            }();

            // 用该条目更新 keydir。
            match result {
                // 正常 put:覆盖该键最新位置
                Ok((key, Some(value_loc))) => keydir.insert(key, value_loc),
                // 墓碑:删除索引
                Ok((key, None)) => keydir.remove(&key),
                // 若在文件末尾发现不完整条目,视为不完整写入并截断文件。
                // 场景:写入中断电可能导致文件末尾只写了一半数据。
                // 处理:假定末尾不完整数据无效,set_len(offset) 截到最后一条完整记录。
                Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => {
                    // 记录截断位置
                    error!("Found incomplete entry at offset {offset}, truncating file");
                    // 截断到最后完整条目末尾
                    self.file.set_len(offset)?; // 截断文件
                    // 结束扫描
                    break;
                // 结束当前作用域
                }
                // 其它 IO 错误向上传播
                Err(err) => return Err(err.into()),
            // 结束当前作用域
            };
        // 结束当前作用域
        }

        // 返回重建的索引
        Ok(keydir)
    // 结束当前作用域
    }

    /// 从日志文件给定位置读取值。
    fn read_value(&mut self, location: ValueLocation) -> Result<Vec<u8>> {
        // 预先分配要读取的空间。
        let mut value: Vec<u8> = vec![0; location.length];
        // SeekFrom::Start(n) 表示从文件开头向后偏移 n 字节。
        // seek 将文件读取指针移到指定位置。
        self.file.seek(SeekFrom::Start(location.offset))?;
        // read_exact 从当前指针读取,直到填满缓冲区(location.length 字节)。
        self.file.read_exact(&mut value)?;
        // 返回读取的值字节
        Ok(value)
    // 结束当前作用域
    }

    /// 向日志追加键值条目;值为 None 表示墓碑。返回条目中值在日志的位置,
    /// 供 [`KeyDir`] 使用。
    fn write_entry(&mut self, key: &[u8], value: Option<&[u8]>) -> Result<ValueLocation> {
        // 本条总字节:8 字节头 + 键 + 值(墓碑无值)
        let length = 8 + key.len() + value.map_or(0, |v| v.len());
        // 追加写:定位到文件末尾,即为本条起始 offset
        let offset = self.file.seek(SeekFrom::End(0))?;
        // 按条目大小分配写缓冲,减少系统调用
        let mut w = BufWriter::with_capacity(length, &mut self.file);

        // 键长度:4 字节 u32。
        w.write_all(&(key.len() as u32).to_be_bytes())?;

        // 值长度:4 字节 i32,墓碑为 -1。
        w.write_all(&value.map_or(-1, |v| v.len() as i32).to_be_bytes())?;

        // 实际键与值。
        w.write_all(key)?;
        // 墓碑时 unwrap_or_default 写空切片
        w.write_all(value.unwrap_or_default())?;
        // 冲刷 BufWriter 到文件
        w.flush()?;

        // 将条目位置转换为值位置。
        // 值从 header+key 之后开始;墓碑 length=0
        Ok(ValueLocation {
            // 列表或字段续项
            offset: offset + 8 + key.len() as u64,
            // 列表或字段续项
            length: value.map_or(0, |v| v.len()),
        // 语法续行
        })
    // 结束当前作用域
    }
// 结束当前作用域
}


// 单元测试(goldenscript 套件)已剥离;见 tests/ 集成测试。