krabka_verified/list_offsets/
selection_model.rs

1#[cfg(creusot)]
2use super::{ListOffsetsSelectionDecision, ListOffsetsSelectionFacts, logic};
3use {
4    super::{
5        ListOffsetsBoundDecision, ListOffsetsBoundFacts, ListOffsetsEarliestFacts,
6        ListOffsetsEpochDecision, ListOffsetsKind,
7    },
8    creusot_std::prelude::*,
9};
10
11/// Classify every timestamp sentinel, including its first supported version.
12#[must_use]
13#[ensures(match result {
14    ListOffsetsKind::Earliest => timestamp@ == -2,
15    ListOffsetsKind::Latest => timestamp@ == -1,
16    ListOffsetsKind::MaxTimestamp => timestamp@ == -3 && version@ >= 7,
17    ListOffsetsKind::EarliestLocal => timestamp@ == -4 && version@ >= 8,
18    ListOffsetsKind::LatestTiered => timestamp@ == -5 && version@ >= 9,
19    ListOffsetsKind::EarliestPendingUpload => timestamp@ == -6 && version@ >= 11,
20    ListOffsetsKind::Timestamp => timestamp@ >= 0,
21    ListOffsetsKind::Unsupported => timestamp@ < -6
22        || (timestamp@ == -6 && version@ < 11)
23        || (timestamp@ == -5 && version@ < 9)
24        || (timestamp@ == -4 && version@ < 8)
25        || (timestamp@ == -3 && version@ < 7),
26})]
27pub const fn list_offsets_kind(timestamp: i64, version: i16) -> ListOffsetsKind {
28    match timestamp {
29        -2 => ListOffsetsKind::Earliest,
30        -1 => ListOffsetsKind::Latest,
31        -3 if version >= 7 => ListOffsetsKind::MaxTimestamp,
32        -4 if version >= 8 => ListOffsetsKind::EarliestLocal,
33        -5 if version >= 9 => ListOffsetsKind::LatestTiered,
34        -6 if version >= 11 => ListOffsetsKind::EarliestPendingUpload,
35        value if value >= 0 => ListOffsetsKind::Timestamp,
36        _ => ListOffsetsKind::Unsupported,
37    }
38}
39
40/// Fence every asserted epoch except the exact `-1` no-epoch sentinel.
41#[must_use]
42#[ensures((result == ListOffsetsEpochDecision::RejectMalformed) == (current_epoch@ < 0))]
43#[ensures((result == ListOffsetsEpochDecision::Proceed) == (current_epoch@ >= 0
44    && (requested_epoch@ == -1 || requested_epoch@ == current_epoch@)))]
45#[ensures((result == ListOffsetsEpochDecision::Fenced) == (current_epoch@ >= 0
46    && requested_epoch@ != -1 && requested_epoch@ < current_epoch@))]
47#[ensures((result == ListOffsetsEpochDecision::Unknown) == (current_epoch@ >= 0
48    && requested_epoch@ > current_epoch@))]
49pub const fn list_offsets_epoch_decision(
50    requested_epoch: i32,
51    current_epoch: i32,
52) -> ListOffsetsEpochDecision {
53    if current_epoch < 0 {
54        ListOffsetsEpochDecision::RejectMalformed
55    } else if requested_epoch == -1 || requested_epoch == current_epoch {
56        ListOffsetsEpochDecision::Proceed
57    } else if requested_epoch < current_epoch {
58        ListOffsetsEpochDecision::Fenced
59    } else {
60        ListOffsetsEpochDecision::Unknown
61    }
62}
63
64/// Select LEO for replicas, HWM for ordinary consumers, and `min(LSO, HWM)`
65/// for read-committed consumers.
66#[must_use]
67#[ensures(match result {
68    ListOffsetsBoundDecision::RejectMalformed => {
69        (facts.replica_id@ != -1 && facts.log_end@ < 0)
70            || (facts.replica_id@ == -1 && facts.high_watermark@ < 0)
71            || (facts.replica_id@ == -1 && facts.isolation_level@ == 1
72                && facts.last_stable@ < 0)
73    }
74    ListOffsetsBoundDecision::Bound { offset } => {
75        offset@ >= 0
76            && (facts.replica_id@ != -1 ==> offset@ == facts.log_end@)
77            && (facts.replica_id@ == -1 && facts.isolation_level@ != 1
78                ==> offset@ == facts.high_watermark@)
79            && (facts.replica_id@ == -1 && facts.isolation_level@ == 1
80                ==> offset@ == if facts.last_stable@ < facts.high_watermark@ {
81                    facts.last_stable@
82                } else {
83                    facts.high_watermark@
84                })
85    }
86})]
87pub const fn list_offsets_bound_decision(facts: ListOffsetsBoundFacts) -> ListOffsetsBoundDecision {
88    if (facts.replica_id != -1 && facts.log_end < 0)
89        || (facts.replica_id == -1 && facts.high_watermark < 0)
90        || (facts.replica_id == -1 && facts.isolation_level == 1 && facts.last_stable < 0)
91    {
92        return ListOffsetsBoundDecision::RejectMalformed;
93    }
94    let offset = if facts.replica_id != -1 {
95        facts.log_end
96    } else if facts.isolation_level == 1 {
97        if facts.last_stable < facts.high_watermark {
98            facts.last_stable
99        } else {
100            facts.high_watermark
101        }
102    } else {
103        facts.high_watermark
104    };
105    ListOffsetsBoundDecision::Bound { offset }
106}
107
108/// Select the equal logical minimum across every available tier.
109#[must_use]
110#[ensures(match result {
111    None => facts.local@ < 0
112        || (facts.has_remote && facts.remote@ < 0)
113        || (facts.has_diskless && facts.diskless@ < 0),
114    Some(offset) => facts.local@ >= 0
115        && (!facts.has_remote || facts.remote@ >= 0)
116        && (!facts.has_diskless || facts.diskless@ >= 0)
117        && offset@ <= facts.local@
118        && (!facts.has_remote || offset@ <= facts.remote@)
119        && (!facts.has_diskless || offset@ <= facts.diskless@)
120        && (offset@ == facts.local@
121            || (facts.has_remote && offset@ == facts.remote@)
122            || (facts.has_diskless && offset@ == facts.diskless@)),
123})]
124pub const fn list_offsets_earliest(facts: ListOffsetsEarliestFacts) -> Option<i64> {
125    if facts.local < 0
126        || (facts.has_remote && facts.remote < 0)
127        || (facts.has_diskless && facts.diskless < 0)
128    {
129        return None;
130    }
131    let local_or_remote = if facts.has_remote && facts.remote < facts.local {
132        facts.remote
133    } else {
134        facts.local
135    };
136    let offset = if facts.has_diskless && facts.diskless < local_or_remote {
137        facts.diskless
138    } else {
139        local_or_remote
140    };
141    Some(offset)
142}
143
144// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
145#[cfg(creusot)]
146#[cfg_attr(test, mutants::skip)]
147#[logic(open)]
148/// The Kafka rule for one partition row's final `ListOffsets` value.
149///
150/// A malformed input fails closed, and a lookup that found nothing (`-1`) is
151/// unknown. `EARLIEST` and `EARLIEST_LOCAL` answer a log start, which Kafka
152/// returns whatever the isolation bound, so they resolve to the candidate
153/// unclamped. `LATEST` answers the isolation bound itself, so it resolves to
154/// the lower of the candidate and `last_fetchable`. Every other kind is
155/// record-derived and, as in `Partition.fetchOffsetForTimestamp`, resolves
156/// only when its offset is strictly below `last_fetchable`; otherwise it is
157/// unknown. A resolved row carries the candidate's timestamp and leader epoch.
158pub fn list_offsets_selection_model(
159    facts: ListOffsetsSelectionFacts,
160) -> ListOffsetsSelectionDecision {
161    pearlite! {
162        if facts.kind == ListOffsetsKind::Unsupported
163            || facts.candidate_offset@ < -1
164            || facts.candidate_epoch@ < -1
165            || facts.last_fetchable@ < 0
166        {
167            ListOffsetsSelectionDecision::RejectMalformed
168        } else if facts.candidate_offset@ == -1 {
169            ListOffsetsSelectionDecision::Unknown
170        } else if facts.kind == ListOffsetsKind::Earliest
171            || facts.kind == ListOffsetsKind::EarliestLocal
172        {
173            ListOffsetsSelectionDecision::Resolved {
174                offset: facts.candidate_offset,
175                timestamp: facts.candidate_timestamp,
176                leader_epoch: facts.candidate_epoch,
177            }
178        } else if facts.kind == ListOffsetsKind::Latest {
179            ListOffsetsSelectionDecision::Resolved {
180                offset: if facts.last_fetchable@ < facts.candidate_offset@ {
181                    facts.last_fetchable
182                } else {
183                    facts.candidate_offset
184                },
185                timestamp: facts.candidate_timestamp,
186                leader_epoch: facts.candidate_epoch,
187            }
188        } else if facts.candidate_offset@ < facts.last_fetchable@ {
189            ListOffsetsSelectionDecision::Resolved {
190                offset: facts.candidate_offset,
191                timestamp: facts.candidate_timestamp,
192                leader_epoch: facts.candidate_epoch,
193            }
194        } else {
195            ListOffsetsSelectionDecision::Unknown
196        }
197    }
198}
krabka_verified/list_offsets/selection_model.rs

krabka_verified/list_offsets/
selection_model.rs
