krabka_verified/retention/
delete_target.rs

1#[cfg(creusot)]
2use super::{remote_prefix_charge, remote_retention_model};
3use {super::RemoteRetentionSegment, creusot_std::prelude::*};
4
5/// Count the oldest finished remote segments that Kafka's remote retention
6/// deletes.
7///
8/// `size_debt` is the total size minus `retention.bytes`, or zero when
9/// `retention.bytes` is unset or not exceeded. A tier that accepts no delete
10/// (`deletes_allowed` false) selects nothing.
11/// A blocked row stops the prefix even when its floor, time or size axis
12/// would otherwise select it.
13///
14/// The result equals `remote_retention_model` from the oldest segment.
15/// That fold states the Kafka rule. Independent bounds justify every selected
16/// row by a floor/time breach or a funded charged prefix, and require the first
17/// kept row to be blocked or fail all three axes. Zero debt cannot delete an
18/// unbreached empty row; an expired oversized row exhausts debt without
19/// restarting size deletion.
20#[ensures(result@ == if deletes_allowed {
21    remote_retention_model(segments@, 0, size_debt@)
22} else {
23    0
24})]
25#[ensures(result@ <= [email protected]())]
26#[ensures(!deletes_allowed ==> result@ == 0)]
27#[ensures(retention_prefix_funded(segments@, size_debt@, result@))]
28#[ensures(deletes_allowed && result@ < [email protected]()
29    ==> segments@[result@].blocked || (!segments@[result@].log_start_breached && !segments@[result@].time_expired
30        && (size_debt@ == 0 || remote_prefix_charge(segments@, result@) >= size_debt@
31            || remote_prefix_charge(segments@, result@ + 1) > size_debt@)))]
32#[must_use]
33pub fn remote_retention_prefix(
34    deletes_allowed: bool,
35    segments: &[RemoteRetentionSegment],
36    size_debt: u64,
37) -> usize {
38    if !deletes_allowed {
39        return 0;
40    }
41    let mut debt = size_debt;
42    let mut len = 0usize;
43    #[invariant(len@ <= [email protected]())]
44    #[invariant(remote_prefix_charge(segments@, len@) >= 0)]
45    #[invariant(debt@ == if remote_prefix_charge(segments@, len@) <= size_debt@ {
46        size_debt@ - remote_prefix_charge(segments@, len@)
47    } else { 0 })]
48    #[invariant(retention_prefix_funded(segments@, size_debt@, len@))]
49    #[invariant(remote_retention_model(segments@, len@, debt@)
50        == remote_retention_model(segments@, 0, size_debt@))]
51    #[variant([email protected]() - len@)]
52    while len < segments.len() {
53        let Some(&segment) = segments.get(len) else {
54            break;
55        };
56        if segment.blocked {
57            break;
58        }
59        if segment.log_start_breached {
60            // Deleted below the floor; the size debt is untouched.
61        } else if segment.time_expired {
62            debt = debt.saturating_sub(segment.size);
63        } else if debt > 0 && segment.size <= debt {
64            debt -= segment.size;
65        } else {
66            break;
67        }
68        len += 1;
69    }
70    len
71}
72
73/// The exclusive delete-through target after an inclusive last offset.
74///
75/// No selected segment gives no target. An inclusive last offset of
76/// `i64::MAX` has no representable successor, so it fails closed with no
77/// target too.
78#[ensures(match (last_offset, result) {
79    (None, result) => result == None,
80    (Some(last), None) => last@ == i64::MAX@,
81    (Some(last), Some(target)) => last@ < i64::MAX@ && target@ == last@ + 1,
82})]
83#[must_use]
84pub fn retention_delete_target(last_offset: Option<i64>) -> Option<i64> {
85    match last_offset {
86        Some(last) if last < i64::MAX => Some(last + 1),
87        Some(_) | None => None,
88    }
89}
90
91// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
92#[cfg(creusot)]
93#[cfg_attr(test, mutants::skip)]
94#[logic(open)]
95/// Each deleted row breaches the floor or time policy, or fits the charged size debt.
96fn retention_prefix_funded(segments: Seq<RemoteRetentionSegment>, debt: Int, count: Int) -> bool {
97    pearlite! { forall<i: Int> 0 <= i && i < count ==> !segments[i].blocked && (segments[i].log_start_breached
98    || segments[i].time_expired || (super::remote_charge_funded(debt, remote_prefix_charge(segments, i), remote_prefix_charge(segments, i + 1)))) }
99}
krabka_verified/retention/delete_target.rs

krabka_verified/retention/
delete_target.rs
