krabka_verified/diskless/
logical_range.rs

1use {
2    super::DisklessRetentionPolicy,
3    crate::log_index::{RangeEntry, upper_range_cursor},
4    creusot_std::prelude::*,
5};
6
7/// The bytes of the oldest `count` ranges.
8// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
9#[cfg(creusot)]
10#[cfg_attr(test, mutants::skip)]
11#[logic]
12#[variant(count)]
13pub fn indexed_bytes(byte_lens: Seq<u64>, count: Int) -> Int {
14    pearlite! {
15        if count <= 0 {
16            0
17        } else {
18            indexed_bytes(byte_lens, count - 1) + byte_lens[count - 1]@
19        }
20    }
21}
22
23/// Kafka's `diff` when the size walk reaches range `i`: the index's bytes
24/// over `retention.bytes`, less the ranges already expired before `i`. It is
25/// negative when the index fits the budget.
26// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
27#[cfg(creusot)]
28#[cfg_attr(test, mutants::skip)]
29#[logic]
30pub fn size_debt_before(byte_lens: Seq<u64>, budget: u64, i: Int) -> Int {
31    pearlite! {
32        indexed_bytes(byte_lens, byte_lens.len()) - budget@ - indexed_bytes(byte_lens, i)
33    }
34}
35
36/// Whether Kafka's `UnifiedLog.deleteOldSegments` lets range `i` expire,
37/// given that every older range expires first.
38///
39/// - `deleteLogStartOffsetBreachedSegments`: the range ends below the
40///   `DeleteRecords` floor.
41/// - `deleteRetentionSizeBreachedSegments`: `diff - segmentSize >= 0`, so
42///   `retention.bytes` never deletes past its own budget.
43/// - `deleteRetentionMsBreachedSegments`: `now - largestTimestamp >
44///   retention.ms`, with a horizon `now - retention.ms` that `i64` cannot
45///   represent expiring nothing.
46// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
47#[cfg(creusot)]
48#[cfg_attr(test, mutants::skip)]
49#[logic]
50pub fn expirable(
51    max_timestamps: Seq<i64>,
52    byte_lens: Seq<u64>,
53    last_offsets: Seq<i64>,
54    policy: DisklessRetentionPolicy,
55    i: Int,
56) -> bool {
57    pearlite! {
58        last_offsets[i]@ < policy.log_start_offset@
59            || match policy.retention_bytes {
60                Some(budget) => size_debt_before(byte_lens, budget, i) - byte_lens[i]@ >= 0,
61                None => false,
62            }
63            || match policy.retention_ms {
64                Some(retention) => i64::MIN@ <= policy.now_ms@ - retention@
65                    && policy.now_ms@ - retention@ <= i64::MAX@
66                    && max_timestamps[i]@ < policy.now_ms@ - retention@,
67                None => false,
68            }
69    }
70}
71
72/// Select the oldest contiguous prefix of one diskless partition's committed
73/// WAL index ranges that retention allows to expire.
74///
75/// Kafka runs three walks oldest first, each stopping at the first segment it
76/// must keep: the log-start-offset walk, then the size walk over what is left,
77/// then the time walk over what is left after that. The first two predicates
78/// only ever hold on a prefix (the size one because sizes are nonnegative, the
79/// floor one because the host supplies ranges in offset order), so the three
80/// walks together expire exactly the longest prefix on which one of the three
81/// holds at every range. That prefix is what this returns, capped one short of
82/// the newest range; see `expirable`.
83///
84/// The ranges arrive oldest first, one entry per index range, and the three
85/// slices are parallel.
86///
87/// The newest range never expires. Kafka keeps the active segment for the same
88/// reason, and here it is also what keeps the flusher's `flushed_frontier`
89/// pointing past the last flushed offset: an empty index would send the next
90/// tick back to the local log start and re-upload a prefix the bucket holds.
91#[requires([email protected]() == [email protected]())]
92#[requires([email protected]() == [email protected]())]
93#[ensures(result@ <= [email protected]())]
94#[ensures([email protected]() > 0 ==> result@ < [email protected]())]
95#[ensures(forall<i: Int> 0 <= i && i < result@ ==>
96    expirable(max_timestamps@, byte_lens@, last_offsets@, policy, i))]
97#[ensures(result@ + 1 < [email protected]() ==>
98    !expirable(max_timestamps@, byte_lens@, last_offsets@, policy, result@))]
99#[must_use]
100pub fn diskless_retention_prefix(
101    max_timestamps: &[i64],
102    byte_lens: &[u64],
103    last_offsets: &[i64],
104    policy: DisklessRetentionPolicy,
105) -> usize {
106    if matches!(max_timestamps.len(), 0) {
107        return 0;
108    }
109    let max_expire = max_timestamps.len() - 1;
110    // Kafka's `diff`, `None` once it is negative. A `u128` holds the sum of
111    // any slice of `u64` lengths without overflow.
112    let mut debt: Option<u128> = 'debt: {
113        let Some(budget) = policy.retention_bytes else {
114            break 'debt None;
115        };
116        let mut indexed = 0u128;
117        let mut scanned = 0usize;
118        #[invariant(scanned@ <= [email protected]())]
119        #[invariant(indexed@ == indexed_bytes(byte_lens@, scanned@))]
120        #[invariant(indexed@ <= scanned@ * u64::MAX@)]
121        #[variant([email protected]() - scanned@)]
122        while scanned < byte_lens.len() {
123            proof_assert!(indexed_bytes(byte_lens@, scanned@ + 1)
124                == indexed_bytes(byte_lens@, scanned@) + byte_lens@[scanned@]@);
125            if let Some(&byte_len) = byte_lens.get(scanned) {
126                indexed += u128::from(byte_len);
127            }
128            scanned += 1;
129        }
130        indexed.checked_sub(u128::from(budget))
131    };
132
133    let mut len = 0usize;
134    #[invariant(len@ <= max_expire@)]
135    #[invariant(match policy.retention_bytes {
136        Some(budget) => match debt {
137            Some(debt) => debt@ == size_debt_before(byte_lens@, budget, len@),
138            None => size_debt_before(byte_lens@, budget, len@) < 0,
139        },
140        None => debt == None,
141    })]
142    #[invariant(forall<i: Int> 0 <= i && i < len@ ==>
143        expirable(max_timestamps@, byte_lens@, last_offsets@, policy, i))]
144    #[variant(max_expire@ - len@)]
145    while len < max_expire {
146        proof_assert!(indexed_bytes(byte_lens@, len@ + 1)
147            == indexed_bytes(byte_lens@, len@) + byte_lens@[len@]@);
148        let (Some(&last_offset), Some(&max_timestamp), Some(&byte_len)) = (
149            last_offsets.get(len),
150            max_timestamps.get(len),
151            byte_lens.get(len),
152        ) else {
153            break;
154        };
155        let below_floor = last_offset < policy.log_start_offset;
156        let aged_out = matches!(
157            policy.retention_ms,
158            Some(retention) if matches!(
159                policy.now_ms.checked_sub(retention),
160                Some(horizon) if max_timestamp < horizon
161            )
162        );
163        let size = u128::from(byte_len);
164        let over_budget = matches!(debt, Some(debt) if debt >= size);
165        if !below_floor && !aged_out && !over_budget {
166            break;
167        }
168        if let Some(remaining) = debt {
169            debt = remaining.checked_sub(size);
170        }
171        len += 1;
172    }
173    len
174}
175
176/// Permit object deletion only after the grace period and with no index
177/// reference in the projection protected by the caller's cache lock.
178#[ensures(result == (!referenced && grace_elapsed))]
179#[must_use]
180pub const fn diskless_object_reclaimable(referenced: bool, grace_elapsed: bool) -> bool {
181    !referenced && grace_elapsed
182}
183
184/// Select the covering logical range, or the first successor after a gap.
185#[requires(forall<i: Int> 0 <= i && i < [email protected]()
186    ==> entries@[i].key@ <= entries@[i].coordinate@)]
187#[requires(forall<i: Int, j: Int> 0 <= i && i < j && j < [email protected]()
188    ==> entries@[i].coordinate@ < entries@[j].key@)]
189#[ensures(match result {
190    Some(index) => index@ < [email protected]()
191        && ((entries@[index@].key@ <= requested@ && requested@ <= entries@[index@].coordinate@)
192            || (requested@ < entries@[index@].key@
193                && index@ > 0
194                && entries@[index@ - 1].coordinate@ < requested@)),
195    None => [email protected]() == 0
196        || requested@ < entries@[0].key@
197        || entries@[[email protected]() - 1].coordinate@ < requested@,
198})]
199#[must_use]
200pub fn diskless_logical_range(entries: &[RangeEntry], requested: i64) -> Option<usize> {
201    let lo = upper_range_cursor(entries, requested);
202    if lo == 0 {
203        return None;
204    }
205    let floor = lo - 1;
206    match entries.get(floor) {
207        Some(entry) if requested <= entry.coordinate => Some(floor),
208        _ if lo < entries.len() => Some(lo),
209        _ => None,
210    }
211}
krabka_verified/diskless/logical_range.rs

krabka_verified/diskless/
logical_range.rs
