krabka_verified/retention/
remote_retention_model.rs

1#[cfg(creusot)]
2use super::{RemoteRetentionSegment, local_prefix_bytes};
3use {super::LocalRetentionSegment, creusot_std::prelude::*};
4
5/// Decide whether one held barrier cut falls outside the retained epoch
6/// window.
7///
8/// A nonpositive retention count and an unrepresentable cutoff both fail
9/// closed by expiring nothing.
10#[ensures(result == (retained_cuts@ > 0
11    && published_epoch@ - retained_cuts@ >= i64::MIN@
12    && held_epoch@ <= published_epoch@ - retained_cuts@))]
13#[must_use]
14pub fn barrier_cut_expired(published_epoch: i64, retained_cuts: i32, held_epoch: i64) -> bool {
15    if retained_cuts <= 0 {
16        return false;
17    }
18    let retained_cuts = i64::from(retained_cuts);
19    if published_epoch < i64::MIN + retained_cuts {
20        return false;
21    }
22    held_epoch <= published_epoch - retained_cuts
23}
24
25// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
26#[cfg(creusot)]
27#[cfg_attr(test, mutants::skip)]
28#[logic(open)]
29/// How many segments the local walk may consider: every one of them, except
30/// a newest segment that is empty. That is Kafka's `deletableSegments`, which
31/// never returns a last segment of size zero (`isLastSegmentAndEmpty`), so a
32/// log never loses the segment it appends to without a record having left
33/// with it.
34pub fn local_retention_limit(segments: Seq<LocalRetentionSegment>) -> Int {
35    pearlite! {
36        if segments.len() > 0 && segments[segments.len() - 1].size@ == 0 {
37            segments.len() - 1
38        } else {
39            segments.len()
40        }
41    }
42}
43
44// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
45#[cfg(creusot)]
46#[cfg_attr(test, mutants::skip)]
47#[logic(open)]
48/// The reference rule for local retention, from segment `i` onwards.
49///
50/// Kafka's `UnifiedLog.deleteOldSegments` runs three passes over the local
51/// log in this order: the log-start-offset breach, then `retention.bytes`,
52/// then `retention.ms`. Each pass deletes an oldest prefix and stops at the
53/// first segment its predicate rejects. The size pass
54/// (`deleteRetentionSizeBreachedSegments`) deletes a segment only while
55/// `diff - segment.size() >= 0`, and it runs only when the log is at least
56/// its budget.
57///
58/// While `sizing` holds, the fold is in the size pass with `debt` bytes left
59/// to reclaim. A segment that does not fit ends the size pass, and the
60/// fold continues through the time pass. The time flag also carries the
61/// log-start breach. That breach covers an oldest prefix, so handling it
62/// inside the time pass deletes the same segments as Kafka's separate first
63/// pass. A breached segment that fits the size debt is charged against it,
64/// exactly as Kafka's size pass sees the log after the breach pass. A
65/// breached segment that does not fit leaves Kafka's recomputed debt
66/// negative, and then Kafka runs no size pass either.
67///
68/// The fold stops at a blocked segment and at `limit`.
69#[variant(limit - i)]
70pub fn local_retention_model(
71    segments: Seq<LocalRetentionSegment>,
72    limit: Int,
73    i: Int,
74    debt: Int,
75    sizing: bool,
76) -> Int {
77    pearlite! {
78        if i >= limit || i >= segments.len() || segments[i].blocked {
79            i
80        } else if sizing && segments[i].size@ <= debt {
81            local_retention_model(segments, limit, i + 1, debt - segments[i].size@, true)
82        } else if segments[i].expired {
83            local_retention_model(segments, limit, i + 1, debt, false)
84        } else {
85            i
86        }
87    }
88}
89
90/// Count the oldest local segments that Kafka's local retention passes
91/// delete.
92///
93/// `segments` is every local segment, oldest first, and the newest one is
94/// the segment the log appends to: Kafka's `deletableSegments` walks the
95/// active segment too. A result that covers every segment tells the host to
96/// roll first and then delete them all, the way Kafka's `deleteSegments`
97/// does (`if (numberOfSegments == numToDelete) roll()`); the walk never
98/// takes a newest segment that is empty, so the roll always leaves a fresh
99/// empty segment behind and never has to recreate the one it deletes.
100///
101/// `size_debt` is `None` when Kafka skips the size pass: `retention.bytes`
102/// is unset, the policy does not delete, or the log is smaller than the
103/// budget. Otherwise it is the log size minus the budget. The walk never
104/// passes a blocked segment.
105///
106/// The result equals `local_retention_model` from the oldest segment,
107/// limited by `local_retention_limit`. That fold states the Kafka rule.
108/// Before that limit, the first kept segment is blocked or not expired.
109/// Every selected non-expired segment fits the initial size debt together with
110/// every older segment. Before the limit, an unblocked kept segment exceeds
111/// that debt. These independent cost bounds reject a selector/reference pair
112/// that omits the size pass or deletes beyond the budget.
113#[ensures(result@ == match size_debt {
114    None => local_retention_model(
115        segments@, local_retention_limit(segments@), 0, 0, false),
116    Some(debt) => local_retention_model(
117        segments@, local_retention_limit(segments@), 0, debt@, true),
118})]
119#[ensures(result@ <= [email protected]())]
120#[ensures(forall<i: Int> 0 <= i && i < result@ ==> !segments@[i].blocked)]
121#[ensures([email protected]() > 0 && segments@[[email protected]() - 1].size@ == 0
122    ==> result@ < [email protected]())]
123#[ensures(result@ < local_retention_limit(segments@)
124    ==> segments@[result@].blocked || !segments@[result@].expired)]
125#[ensures(forall<i: Int> 0 <= i && i < result@ ==> segments@[i].expired
126    || match size_debt { None => false, Some(debt) => local_prefix_bytes(segments@, i + 1) <= debt@ })]
127#[ensures(result@ < local_retention_limit(segments@) && !segments@[result@].blocked
128    ==> match size_debt { None => true, Some(debt) => local_prefix_bytes(segments@, result@ + 1) > debt@ })]
129#[must_use]
130pub fn local_retention_prefix(segments: &[LocalRetentionSegment], size_debt: Option<u64>) -> usize {
131    let limit = match segments.len().checked_sub(1) {
132        Some(newest) if matches!(segments.get(newest), Some(segment) if segment.size == 0) => {
133            newest
134        }
135        _ => segments.len(),
136    };
137    let (mut debt, mut sizing) = (0, false);
138    if let Some(initial) = size_debt {
139        (debt, sizing) = (initial, true);
140    }
141    let mut len = 0usize;
142    #[invariant(len@ <= limit@)]
143    #[invariant(forall<i: Int> 0 <= i && i < len@ ==> !segments@[i].blocked)]
144    #[invariant(local_prefix_bytes(segments@, len@) >= 0)]
145    #[invariant(match size_debt {
146        None => !sizing,
147        Some(initial) => if sizing {
148            local_prefix_bytes(segments@, len@) + debt@ == initial@
149        } else { local_prefix_bytes(segments@, len@) > initial@ },
150    })]
151    #[invariant(forall<i: Int> 0 <= i && i < len@ ==> segments@[i].expired
152        || match size_debt { None => false, Some(initial) => local_prefix_bytes(segments@, i + 1) <= initial@ })]
153    #[invariant(limit@ == local_retention_limit(segments@))]
154    #[invariant(local_retention_model(segments@, limit@, len@, debt@, sizing) == match size_debt {
155        None => local_retention_model(segments@, limit@, 0, 0, false),
156        Some(initial) => local_retention_model(segments@, limit@, 0, initial@, true),
157    })]
158    #[variant(limit@ - len@)]
159    while len < limit {
160        let Some(&segment) = segments.get(len) else {
161            break;
162        };
163        if segment.blocked {
164            break;
165        }
166        if sizing && segment.size <= debt {
167            debt -= segment.size;
168        } else if segment.expired {
169            sizing = false;
170        } else {
171            break;
172        }
173        len += 1;
174    }
175    len
176}
177
178// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
179#[cfg(creusot)]
180#[cfg_attr(test, mutants::skip)]
181#[logic(open)]
182/// The reference rule for remote retention, from segment `i` onwards, with
183/// `debt` bytes of `retention.bytes` breach left to reclaim.
184///
185/// This is Kafka's `RemoteLogManager.cleanupExpiredRemoteLogSegments`, which
186/// asks `RemoteLogRetentionHandler` about each segment in turn and stops at
187/// the first one it keeps. A log-start breach deletes the segment and leaves
188/// the debt alone. A `retention.ms` breach deletes it and lowers the debt by
189/// its size, but not below zero. Otherwise the segment goes only when the
190/// debt is positive and still covers the whole segment
191/// (`isSegmentBreachedByRetentionSize`).
192#[variant(segments.len() - i)]
193pub fn remote_retention_model(segments: Seq<RemoteRetentionSegment>, i: Int, debt: Int) -> Int {
194    pearlite! {
195        if i >= segments.len() || segments[i].blocked {
196            i
197        } else if segments[i].log_start_breached {
198            remote_retention_model(segments, i + 1, debt)
199        } else if segments[i].time_expired {
200            remote_retention_model(
201                segments,
202                i + 1,
203                if segments[i].size@ <= debt { debt - segments[i].size@ } else { 0 },
204            )
205        } else if debt > 0 && segments[i].size@ <= debt {
206            remote_retention_model(segments, i + 1, debt - segments[i].size@)
207        } else {
208            i
209        }
210    }
211}
krabka_verified/retention/remote_retention_model.rs

krabka_verified/retention/
remote_retention_model.rs
