krabka_log/
leader_epoch_checkpoint.rs

1//! Per-partition `.leader-epoch-checkpoint` file. The two-column text format
2//! matches Apache Kafka exactly:
3//!
4//! ```text
5//!   0          <-- header version
6//!   <n>        <-- row count
7//!   <epoch_0> <start_offset_0>
8//!   <epoch_1> <start_offset_1>
9//!   ...
10//! ```
11//!
12//! The byte layout stays the same, so `kafka-dump-log` can read these files.
13
14pub use krabka_verified::{EpochEntry, epoch_and_offset_for_entries};
15use {
16    crate::io::LogIo,
17    krabka_ids::{LeaderEpoch, Offset},
18    std::{path::PathBuf, sync::Arc},
19};
20
21mod file;
22mod lookup;
23mod mutation;
24
25#[cfg(test)]
26mod fuzz;
27#[cfg(test)]
28mod test_support;
29
30#[derive(Debug)]
31pub struct LeaderEpochCheckpoint {
32    path: PathBuf,
33    io: Arc<dyn LogIo>,
34    entries: Vec<EpochEntry>,
35}
36
37/// Kafka sentinel: "no leader epoch information".
38#[cfg(test)]
39pub const UNDEFINED_EPOCH: LeaderEpoch = LeaderEpoch(-1);
40/// Kafka sentinel: "no offset".
41pub const UNDEFINED_OFFSET: Offset = Offset(-1);
42
43impl LeaderEpochCheckpoint {
44    /// Route this checkpoint's writes, syncs and renames through `io`.
45    ///
46    /// Only [`crate::Log::test_set_io`] calls this: the checkpoint is opened
47    /// with the real filesystem and stays there unless a test replaces it.
48    #[cfg(any(test, feature = "test-helpers"))]
49    pub(crate) fn set_io(&mut self, io: Arc<dyn LogIo>) { self.io = io; }
50
51    #[must_use]
52    pub fn latest_epoch(&self) -> Option<LeaderEpoch> { self.entries.iter().map(|e| e.epoch).max() }
53
54    #[must_use]
55    pub fn entries(&self) -> &[EpochEntry] { &self.entries }
56}
57
58pub fn is_strict_successor(previous: &EpochEntry, next: &EpochEntry) -> bool {
59    previous.epoch < next.epoch && previous.start_offset < next.start_offset
60}
61
62/// Pure core of [`LeaderEpochCheckpoint::append`]: an idempotent
63/// push-if-absent. It returns `true` when it added a new entry, so the caller
64/// knows that it must flush.
65pub fn append_to(entries: &mut Vec<EpochEntry>, epoch: LeaderEpoch, start_offset: Offset) -> bool {
66    if entries.iter().any(|e| e.epoch == epoch) {
67        return false;
68    }
69    entries.push(EpochEntry {
70        epoch,
71        start_offset,
72    });
73    true
74}
75
76/// Pure core of [`LeaderEpochCheckpoint::truncate_from_end`]: drop entries that
77/// begin at or after `end_offset`.
78pub fn truncate_to(entries: &mut Vec<EpochEntry>, end_offset: Offset) {
79    entries.retain(|e| e.start_offset < end_offset);
80}
81
82#[cfg(test)]
83#[path = "leader_epoch_model.rs"]
84mod leader_epoch_model;
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89
90    #[test]
91    fn strict_successor_requires_strictly_increasing_epoch_and_offset() {
92        let e1 = EpochEntry {
93            epoch: LeaderEpoch(1),
94            start_offset: Offset(0),
95        };
96        let e2 = EpochEntry {
97            epoch: LeaderEpoch(1),
98            start_offset: Offset(5),
99        };
100        assert2::assert!(!is_strict_successor(&e1, &e2));
101        assert2::assert!(!is_strict_successor(&e1, &e1));
102        let e3 = EpochEntry {
103            epoch: LeaderEpoch(2),
104            start_offset: Offset(0),
105        };
106        assert2::assert!(!is_strict_successor(&e1, &e3));
107        let e4 = EpochEntry {
108            epoch: LeaderEpoch(2),
109            start_offset: Offset(5),
110        };
111        assert2::assert!(is_strict_successor(&e1, &e4));
112    }
113}
krabka_log/leader_epoch_checkpoint.rs

krabka_log/
leader_epoch_checkpoint.rs
