krabka_log/leader_epoch_checkpoint/
lookup.rs

1//! The read-side queries over a recorded epoch history: where an epoch ends,
2//! which epoch owned an offset, and the KIP-320 `(found_epoch, end_offset)`
3//! pair a follower or consumer truncates to. The pure core over a raw slice
4//! sits here with the methods that delegate to it, so that the divergence
5//! contract is stated once.
6
7#[cfg(test)]
8use super::UNDEFINED_EPOCH;
9use {
10    super::{LeaderEpochCheckpoint, UNDEFINED_OFFSET, epoch_and_offset_for_entries},
11    krabka_ids::{LeaderEpoch, Offset},
12};
13
14impl LeaderEpochCheckpoint {
15    /// End offset of a *recorded* `epoch`. It is the `start_offset` of the
16    /// next-larger recorded epoch, or `log_end_offset` if `epoch` is the
17    /// current epoch. The method returns -1, which is `UNDEFINED_OFFSET`, if
18    /// `epoch` is not recorded.
19    ///
20    /// This is an exact-match lookup for the log's own bookkeeping, not
21    /// Kafka's `endOffsetFor`: a gap epoch or one below the first entry is
22    /// unknown here. Wire answers use [`Self::epoch_and_offset_for`].
23    #[must_use]
24    pub fn end_offset_for_epoch(&self, epoch: LeaderEpoch, log_end_offset: Offset) -> Offset {
25        if !self.entries.iter().any(|e| e.epoch == epoch) {
26            return UNDEFINED_OFFSET;
27        }
28        // End of `epoch` is the start of the next-larger epoch. Higher
29        // epochs always carry higher start offsets, so the minimum start
30        // among epochs `> epoch` is that next epoch's start; if `epoch` is
31        // the latest, there is none and the end is the log end. No clone or
32        // sort — a single linear pass.
33        self.entries
34            .iter()
35            .filter(|e| e.epoch > epoch)
36            .map(|e| e.start_offset)
37            .min()
38            .unwrap_or(log_end_offset)
39    }
40
41    /// Floor lookup: return the epoch of the entry whose `start_offset` is
42    /// the greatest value `<= offset`, that is, the leader epoch that owned
43    /// `offset`.
44    ///
45    /// The method returns `None` when the checkpoint has no entries, and when
46    /// `offset` comes before the first entry's `start_offset`. In the second
47    /// case the offset predates every recorded epoch boundary.
48    ///
49    /// Entries are stored in increasing `start_offset` order by construction,
50    /// because `append` always writes the current epoch, whose `start_offset`
51    /// is `>=` that of every prior entry. This lookup is therefore a single
52    /// linear scan from the back. It is equivalent to a search for the last
53    /// entry with `start_offset <= offset`.
54    #[must_use]
55    pub fn epoch_for_offset(&self, offset: Offset) -> Option<LeaderEpoch> {
56        self.entries
57            .iter()
58            .filter(|e| e.start_offset <= offset)
59            .max_by_key(|e| e.start_offset)
60            .map(|e| e.epoch)
61    }
62
63    /// Kafka `LeaderEpochFileCache.endOffsetFor`. It returns
64    /// `(found_epoch, end_offset)`: the epoch that the requested offset range
65    /// really belongs to on this log, and the first offset *after* that epoch.
66    /// The broker uses it to detect follower and consumer log divergence
67    /// (KIP-101/320). The Creusot contract of
68    /// [`epoch_and_offset_for_entries`] pins Kafka's whole case table:
69    ///
70    ///  - `requested == UNDEFINED_EPOCH`       → `(UNDEFINED_EPOCH, UNDEFINED_OFFSET)`
71    ///  - `requested == latest recorded epoch` → `(requested, log_end_offset)`
72    ///  - no recorded epoch above `requested`  → `(UNDEFINED_EPOCH, UNDEFINED_OFFSET)`
73    ///    (this covers an empty checkpoint and a future epoch)
74    ///  - `requested` below every recorded epoch → `(requested, first_recorded_start)`
75    ///  - otherwise (gap or exact older match)  → `(floor_epoch, next_epoch_start)`
76    ///
77    /// where `floor_epoch` is the largest recorded epoch `<= requested` and
78    /// `next_epoch_start` is the start of the least recorded epoch
79    /// `> requested`. `(-1, -1)` means "cannot place this epoch": an
80    /// `OffsetForLeaderEpoch` row carries it as is, and a `Fetch` answers it
81    /// with `OFFSET_OUT_OF_RANGE` (`Partition.readRecords`).
82    #[must_use]
83    pub fn epoch_and_offset_for(
84        &self,
85        requested_epoch: LeaderEpoch,
86        log_end_offset: Offset,
87    ) -> (LeaderEpoch, Offset) {
88        epoch_and_offset_for_entries(&self.entries, requested_epoch, log_end_offset)
89    }
90}
91
92#[cfg(test)]
93mod tests {
94    use {
95        super::*,
96        crate::{
97            leader_epoch_checkpoint::{
98                EpochEntry,
99                test_support::{TempCheckpoint, checkpoint, entry, fresh},
100            },
101            test_support::TempPath,
102        },
103        assert2::check,
104    };
105
106    #[test]
107    fn end_offset_for_current_epoch_returns_log_end_offset() {
108        let TempCheckpoint {
109            dir: _d,
110            checkpoint: c,
111        } = checkpoint(&[entry(0, 0), entry(1, 50)]);
112        assert2::assert!(c.end_offset_for_epoch(LeaderEpoch(1), Offset(100)) == 100);
113    }
114
115    #[test]
116    fn end_offset_for_older_epoch_returns_next_start() {
117        let TempCheckpoint {
118            dir: _d,
119            checkpoint: c,
120        } = checkpoint(&[entry(0, 0), entry(1, 50), entry(2, 100)]);
121        assert2::assert!(c.end_offset_for_epoch(LeaderEpoch(0), Offset(200)) == Offset(50));
122        assert2::assert!(c.end_offset_for_epoch(LeaderEpoch(1), Offset(200)) == Offset(100));
123    }
124
125    #[test]
126    fn end_offset_for_unknown_epoch_returns_undefined() {
127        let TempCheckpoint {
128            dir: _d,
129            checkpoint: c,
130        } = checkpoint(&[entry(0, 0)]);
131        assert2::assert!(c.end_offset_for_epoch(LeaderEpoch(7), Offset(200)) == -1);
132    }
133
134    // ── epoch_for_offset ──────────────────────────────────────────────────────
135
136    #[yare::parameterized(
137        zero = { "zero", 0 },
138        positive = { "positive", 100 },
139    )]
140    fn epoch_for_offset_empty_returns_none(_name: &str, offset: i64) {
141        let TempPath { dir: _d, path } = fresh();
142        let c = LeaderEpochCheckpoint::open(path).unwrap();
143        assert2::assert!(c.epoch_for_offset(Offset(offset)) == None);
144    }
145
146    #[test]
147    fn epoch_for_offset_before_first_entry_returns_none() {
148        // Epoch 0 starts at offset 10 (first entry does not start at 0).
149        let TempCheckpoint {
150            dir: _d,
151            checkpoint: c,
152        } = checkpoint(&[entry(0, 10), entry(1, 50)]);
153        assert2::assert!(c.epoch_for_offset(Offset(9)) == None);
154    }
155
156    #[yare::parameterized(
157        start_epoch_0 = { 0, Some(LeaderEpoch(0)), "start of epoch 0" },
158        middle_epoch_0 = { 25, Some(LeaderEpoch(0)), "middle of epoch 0" },
159        last_before_epoch_1 = { 49, Some(LeaderEpoch(0)), "last offset before epoch 1" },
160        start_epoch_1 = { 50, Some(LeaderEpoch(1)), "start of epoch 1" },
161        middle_epoch_1 = { 75, Some(LeaderEpoch(1)), "middle of epoch 1" },
162        last_before_epoch_2 = { 99, Some(LeaderEpoch(1)), "last offset before epoch 2" },
163    )]
164    fn epoch_for_offset_within_epoch_range(offset: i64, want: Option<LeaderEpoch>, why: &str) {
165        let TempCheckpoint {
166            dir: _d,
167            checkpoint: c,
168        } = checkpoint(&[entry(0, 0), entry(1, 50), entry(2, 100)]);
169        check!(
170            c.epoch_for_offset(Offset(offset)) == want,
171            "{why} (offset={offset})"
172        );
173    }
174
175    /// Epoch 1 owns its first offset and all following offsets once it is the
176    /// last recorded epoch.
177    #[yare::parameterized(
178        epoch_boundary = { "epoch 1 start", 50 },
179        past_last = { "past last entry", 100 },
180        far_past_last = { "far past last entry", 999 },
181    )]
182    fn epoch_for_offset_at_or_past_last_entry(_name: &str, offset: i64) {
183        let TempCheckpoint {
184            dir: _d,
185            checkpoint: c,
186        } = checkpoint(&[entry(0, 0), entry(1, 50)]);
187        assert2::assert!(c.epoch_for_offset(Offset(offset)) == Some(LeaderEpoch(1)));
188    }
189
190    #[test]
191    fn epoch_for_offset_single_entry_at_zero() {
192        let TempCheckpoint {
193            dir: _d,
194            checkpoint: c,
195        } = checkpoint(&[entry(0, 0)]);
196        assert2::assert!(c.epoch_for_offset(Offset(0)) == Some(LeaderEpoch(0)));
197        assert2::assert!(c.epoch_for_offset(Offset(1000)) == Some(LeaderEpoch(0)));
198    }
199
200    // ── epoch_and_offset_for (KIP-320) ────────────────────────────────────────
201
202    /// One row per branch of Kafka's `endOffsetFor`, over checkpoints built
203    /// through `append` as the log builds them.
204    #[yare::parameterized(
205        empty = { "empty checkpoint", &[][..], 0, 9, (UNDEFINED_EPOCH, UNDEFINED_OFFSET) },
206        undefined = {
207            "undefined requested epoch",
208            &[entry(0, 0), entry(1, 50)][..],
209            -1,
210            100,
211            (UNDEFINED_EPOCH, UNDEFINED_OFFSET),
212        },
213        latest = {
214            "latest recorded epoch keeps the log end",
215            &[entry(0, 0), entry(1, 50)][..],
216            1,
217            100,
218            (LeaderEpoch(1), Offset(100)),
219        },
220        oldest = {
221            "oldest epoch ends at the next start",
222            &[entry(0, 0), entry(1, 50), entry(2, 100)][..],
223            0,
224            200,
225            (LeaderEpoch(0), Offset(50)),
226        },
227        middle = {
228            "middle epoch ends at the next start",
229            &[entry(0, 0), entry(1, 50), entry(2, 100)][..],
230            1,
231            200,
232            (LeaderEpoch(1), Offset(100)),
233        },
234        gap = {
235            "gap epoch resolves to the floor epoch",
236            &[entry(0, 0), entry(5, 100)][..],
237            3,
238            200,
239            (LeaderEpoch(0), Offset(100)),
240        },
241        future = {
242            "future epoch cannot be placed",
243            &[entry(0, 0), entry(1, 50)][..],
244            7,
245            100,
246            (UNDEFINED_EPOCH, UNDEFINED_OFFSET),
247        },
248        below_first = {
249            "below every recorded epoch keeps the requested epoch",
250            &[entry(3, 30), entry(4, 40)][..],
251            1,
252            100,
253            (LeaderEpoch(1), Offset(30)),
254        },
255    )]
256    fn epoch_and_offset_for_follows_kafka_end_offset_for(
257        name: &str,
258        recorded: &[EpochEntry],
259        requested: i32,
260        log_end: i64,
261        expected: (LeaderEpoch, Offset),
262    ) {
263        let TempCheckpoint {
264            dir: _d,
265            checkpoint: c,
266        } = checkpoint(recorded);
267        check!(
268            c.epoch_and_offset_for(LeaderEpoch(requested), Offset(log_end)) == expected,
269            "{name}"
270        );
271    }
272}
krabka_log/leader_epoch_checkpoint/lookup.rs

krabka_log/leader_epoch_checkpoint/
lookup.rs
