krabka_verified/
checkpoint.rs

1//! `KRaft` checkpoint selection and retention decisions.
2
3use creusot_std::prelude::ensures;
4#[cfg(creusot)]
5use creusot_std::prelude::{Int, invariant};
6
7/// Whether `candidate` is lexicographically newer than `current`.
8#[ensures(result == (candidate_end@ > current_end@
9    || (candidate_end@ == current_end@ && candidate_epoch@ > current_epoch@)))]
10#[must_use]
11pub fn checkpoint_id_newer(
12    candidate_end: i64,
13    candidate_epoch: i32,
14    current_end: i64,
15    current_epoch: i32,
16) -> bool {
17    candidate_end > current_end || (candidate_end == current_end && candidate_epoch > current_epoch)
18}
19
20/// Return the index of the maximum `(end offset, epoch)` checkpoint id.
21#[ensures(match result {
22    Some(selected) => selected@ < [email protected]()
23        && (forall<i: Int> 0 <= i && i < [email protected]()
24            ==> ids@[selected@].0@ > ids@[i].0@
25                || (ids@[selected@].0@ == ids@[i].0@
26                    && ids@[selected@].1@ >= ids@[i].1@)),
27    None => [email protected]() == 0,
28})]
29#[must_use]
30pub fn latest_checkpoint_index(ids: &[(i64, i32)]) -> Option<usize> {
31    if matches!(ids.len(), 0) {
32        return None;
33    }
34    let mut selected = 0usize;
35    let mut i = 1usize;
36    #[invariant(1 <= i@ && i@ <= [email protected]())]
37    #[invariant(selected@ < i@)]
38    #[invariant(forall<k: Int> 0 <= k && k < i@
39        ==> ids@[selected@].0@ > ids@[k].0@
40            || (ids@[selected@].0@ == ids@[k].0@
41                && ids@[selected@].1@ >= ids@[k].1@))]
42    #[variant([email protected]() - i@)]
43    while i < ids.len() {
44        if let (Some(&(offset, epoch)), Some(&(selected_offset, selected_epoch))) =
45            (ids.get(i), ids.get(selected))
46            && checkpoint_id_newer(offset, epoch, selected_offset, selected_epoch)
47        {
48            selected = i;
49        }
50        i += 1;
51    }
52    Some(selected)
53}
54
55/// Retain a checkpoint whose full `(end offset, epoch)` id equals either of
56/// two given ids, and nothing else: no partial match of either survives.
57///
58/// That the two ids are the newest and the runner-up is a host fact, not part
59/// of this contract: `recent_checkpoint_ids` in the controller picks them with
60/// [`latest_checkpoint_index`], applied twice.
61///
62/// Kafka keeps the previous checkpoint alive purely through retention
63/// (`KafkaMetadataLog.cleanSnapshots` never deletes the latest snapshot and
64/// expires older ones only under the metadata retention caps), which is what
65/// lets a follower whose `FetchSnapshot` is mid-flight finish the id it
66/// started on when the leader rolls to a newer one. There is no reference
67/// count: the host passes the runner-up id as `previous`, and for a directory
68/// holding a single checkpoint it passes the selected id as both.
69#[ensures(result == ((candidate_end@ == selected_end@
70        && candidate_epoch@ == selected_epoch@)
71    || (candidate_end@ == previous_end@ && candidate_epoch@ == previous_epoch@)))]
72#[must_use]
73pub fn checkpoint_id_retained(
74    candidate_end: i64,
75    candidate_epoch: i32,
76    selected_end: i64,
77    selected_epoch: i32,
78    previous_end: i64,
79    previous_epoch: i32,
80) -> bool {
81    (candidate_end == selected_end && candidate_epoch == selected_epoch)
82        || (candidate_end == previous_end && candidate_epoch == previous_epoch)
83}
84
85#[cfg(test)]
86mod tests {
87    use super::{checkpoint_id_retained, latest_checkpoint_index};
88
89    #[test]
90    fn latest_index_covers_empty_ties_and_lexicographic_order() {
91        assert2::check!(latest_checkpoint_index(&[]) == None);
92        assert2::check!(latest_checkpoint_index(&[(10, 2)]) == Some(0));
93        assert2::check!(latest_checkpoint_index(&[(10, 2), (10, 9), (11, 1)]) == Some(2));
94        assert2::check!(latest_checkpoint_index(&[(10, 5), (10, 2)]) == Some(0));
95        assert2::check!(latest_checkpoint_index(&[(11, 1), (11, 1), (10, 99)]) == Some(0));
96    }
97
98    #[test]
99    fn retention_keeps_exactly_the_two_given_full_ids() {
100        // Latest (11, 3), previous (10, 3): both survive in full, and neither
101        // a stale id nor a partial match of either does.
102        assert2::check!(checkpoint_id_retained(11, 3, 11, 3, 10, 3));
103        assert2::check!(checkpoint_id_retained(10, 3, 11, 3, 10, 3));
104        assert2::check!(!checkpoint_id_retained(9, 3, 11, 3, 10, 3));
105        assert2::check!(!checkpoint_id_retained(11, 2, 11, 3, 10, 3));
106        assert2::check!(!checkpoint_id_retained(10, 2, 11, 3, 10, 3));
107
108        // A lone checkpoint: the caller passes the selected id as both, so
109        // retention collapses back to that single id.
110        assert2::check!(checkpoint_id_retained(11, 3, 11, 3, 11, 3));
111        assert2::check!(!checkpoint_id_retained(10, 3, 11, 3, 11, 3));
112    }
113}
krabka_verified/checkpoint.rs

krabka_verified/
checkpoint.rs
