krabka_verified/broker/
trim_admission.rs

1use {
2    super::{DeleteRecordsTrimApplication, DeleteRecordsTrimDecision, DeleteRecordsTrimFacts},
3    creusot_std::prelude::*,
4};
5
6/// Admit a trim and cap it at every logical deletion frontier.
7///
8/// `-1` means the current high watermark, which is always admitted. An
9/// explicit request above the high watermark is refused
10/// `RejectOutOfRange`, matching `UnifiedLog.maybeIncrementLogStartOffset`
11/// on Kafka trunk: an explicit offset never enters the uncommitted tail,
12/// even when it is still below the log end offset. A scheduled topic adds
13/// its delivery watermark as a second cap on top of the high watermark.
14/// Stale and repeated requests return the current start and never move it
15/// backwards.
16#[must_use]
17#[ensures({
18    let malformed = facts.requested@ < -1
19        || facts.current_start@ < 0
20        || facts.high_watermark@ < facts.current_start@
21        || facts.log_end@ < facts.high_watermark@
22        || (facts.has_delivery_watermark
23            && facts.delivery_watermark@ < facts.current_start@);
24    let out_of_range = !malformed
25        && facts.requested@ != -1
26        && facts.requested@ > facts.high_watermark@;
27    let resolved = if facts.requested@ == -1 {
28        facts.high_watermark@
29    } else {
30        facts.requested@
31    };
32    let bounded = if facts.has_delivery_watermark
33        && facts.delivery_watermark@ < resolved
34    {
35        facts.delivery_watermark@
36    } else {
37        resolved
38    };
39    match result {
40        DeleteRecordsTrimDecision::RejectMalformed => malformed,
41        DeleteRecordsTrimDecision::RejectOutOfRange => out_of_range,
42        DeleteRecordsTrimDecision::Noop { frontier } => {
43            !malformed && !out_of_range
44                && bounded <= facts.current_start@
45                && frontier@ == facts.current_start@
46        }
47        DeleteRecordsTrimDecision::Apply { frontier } => {
48            !malformed && !out_of_range
49                && bounded > facts.current_start@
50                && frontier@ == bounded
51                && frontier@ <= facts.high_watermark@
52                && frontier@ <= facts.log_end@
53                && (!facts.has_delivery_watermark
54                    || frontier@ <= facts.delivery_watermark@)
55        }
56    }
57})]
58pub const fn delete_records_trim_decision(
59    facts: DeleteRecordsTrimFacts,
60) -> DeleteRecordsTrimDecision {
61    if facts.requested < -1
62        || facts.current_start < 0
63        || facts.high_watermark < facts.current_start
64        || facts.log_end < facts.high_watermark
65        || (facts.has_delivery_watermark && facts.delivery_watermark < facts.current_start)
66    {
67        return DeleteRecordsTrimDecision::RejectMalformed;
68    }
69    if facts.requested != -1 && facts.requested > facts.high_watermark {
70        return DeleteRecordsTrimDecision::RejectOutOfRange;
71    }
72    let resolved = if facts.requested == -1 {
73        facts.high_watermark
74    } else {
75        facts.requested
76    };
77    let bounded = if facts.has_delivery_watermark {
78        if resolved < facts.delivery_watermark {
79            resolved
80        } else {
81            facts.delivery_watermark
82        }
83    } else {
84        resolved
85    };
86    if bounded <= facts.current_start {
87        DeleteRecordsTrimDecision::Noop {
88            frontier: facts.current_start,
89        }
90    } else {
91        DeleteRecordsTrimDecision::Apply { frontier: bounded }
92    }
93}
94
95/// Choose one monotonic trim step, with WAL ordered before the local log.
96///
97/// Re-evaluating this function after a failed step is retry-safe: the chosen
98/// frontier is the maximum of the request and both observed frontiers, so no
99/// retry can regress either store. Completion requires exact equality.
100#[must_use]
101#[ensures({
102    let frontier = if requested@ > wal_start@ {
103        if requested@ > local_start@ { requested@ } else { local_start@ }
104    } else if wal_start@ > local_start@ {
105        wal_start@
106    } else {
107        local_start@
108    };
109    match result {
110        DeleteRecordsTrimApplication::RejectMalformed => {
111            requested@ < 0 || wal_start@ < 0 || local_start@ < 0
112        }
113        DeleteRecordsTrimApplication::TrimWal { frontier: next } => {
114            requested@ >= 0 && wal_start@ >= 0 && local_start@ >= 0
115                && wal_start@ < frontier && next@ == frontier
116        }
117        DeleteRecordsTrimApplication::TrimLocal { frontier: next } => {
118            requested@ >= 0 && wal_start@ >= 0 && local_start@ >= 0
119                && wal_start@ == frontier && local_start@ < frontier
120                && next@ == frontier
121        }
122        DeleteRecordsTrimApplication::Complete { frontier: done } => {
123            requested@ >= 0 && wal_start@ >= 0 && local_start@ >= 0
124                && wal_start@ == frontier && local_start@ == frontier
125                && done@ == frontier
126        }
127    }
128})]
129pub const fn delete_records_trim_application(
130    requested: i64,
131    wal_start: i64,
132    local_start: i64,
133) -> DeleteRecordsTrimApplication {
134    if requested < 0 || wal_start < 0 || local_start < 0 {
135        return DeleteRecordsTrimApplication::RejectMalformed;
136    }
137    let request_or_wal = if requested > wal_start {
138        requested
139    } else {
140        wal_start
141    };
142    let frontier = if request_or_wal > local_start {
143        request_or_wal
144    } else {
145        local_start
146    };
147    if wal_start < frontier {
148        DeleteRecordsTrimApplication::TrimWal { frontier }
149    } else if local_start < frontier {
150        DeleteRecordsTrimApplication::TrimLocal { frontier }
151    } else {
152        DeleteRecordsTrimApplication::Complete { frontier }
153    }
154}
155
156/// Non-negative KIP-932 backlog above the effective share start offset.
157// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
158#[cfg(creusot)]
159#[cfg_attr(test, mutants::skip)]
160#[logic]
161#[cfg_attr(test, mutants::skip)]
162pub fn effective_share_backlog_model(hwm: i64, spso: i64, log_start: i64) -> Int {
163    pearlite! {
164        let base = if spso@ >= 0 && spso@ > log_start@ { spso@ } else { log_start@ };
165        let difference = hwm@ - base;
166        if difference <= 0 {
167            0
168        } else if difference > 9223372036854775807 {
169            9223372036854775807
170        } else {
171            difference
172        }
173    }
174}
175
176#[ensures(result@ == effective_share_backlog_model(hwm, spso, log_start))]
177#[must_use]
178pub fn effective_share_backlog(hwm: i64, spso: i64, log_start: i64) -> i64 {
179    let base = if spso >= 0 && spso > log_start {
180        spso
181    } else {
182        log_start
183    };
184    let difference = hwm.saturating_sub(base);
185    if difference > 0 { difference } else { 0 }
186}
krabka_verified/broker/trim_admission.rs

krabka_verified/broker/
trim_admission.rs
