krabka_verified/
timestamp.rs

1//! Timestamp-scan selection and progress kernels.
2
3use creusot_std::prelude::ensures;
4#[cfg(creusot)]
5use creusot_std::prelude::{Int, Seq, invariant, logic};
6
7// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
8#[cfg(creusot)]
9#[cfg_attr(test, mutants::skip)]
10#[logic(open)]
11/// The complete first-match guarantee shared by direct and indexed scans.
12pub(crate) fn first_timestamp_match(
13    timestamps: Seq<i64>,
14    target: Int,
15    matched: Option<usize>,
16) -> bool {
17    pearlite! {
18        match matched {
19            Some(index) => index@ < timestamps.len()
20                && timestamps[index@]@ >= target
21                && forall<i: Int> 0 <= i && i < index@ ==> timestamps[i]@ < target,
22            None => forall<i: Int> 0 <= i && i < timestamps.len() ==> timestamps[i]@ < target,
23        }
24    }
25}
26
27/// Select the first timestamp at or after `target`.
28#[ensures(first_timestamp_match(timestamps@, target@, result))]
29#[must_use]
30pub fn first_timestamp_index(timestamps: &[i64], target: i64) -> Option<usize> {
31    let mut index = 0usize;
32    #[invariant(index@ <= [email protected]())]
33    #[invariant(forall<i: Int> 0 <= i && i < index@ ==> timestamps@[i]@ < target@)]
34    #[variant([email protected]() - index@)]
35    while index < timestamps.len() {
36        if matches!(timestamps.get(index), Some(&timestamp) if timestamp >= target) {
37            return Some(index);
38        }
39        index += 1;
40    }
41    None
42}
43
44/// Select the earliest occurrence of the maximum timestamp.
45#[ensures(match result {
46    Some(index) => index@ < [email protected]()
47        && forall<i: Int> 0 <= i && i < [email protected]() ==>
48            timestamps@[i]@ <= timestamps@[index@]@
49        && forall<i: Int> 0 <= i && i < index@ ==>
50            timestamps@[i]@ < timestamps@[index@]@,
51    None => [email protected]() == 0,
52})]
53#[must_use]
54pub fn earliest_max_timestamp_index(timestamps: &[i64]) -> Option<usize> {
55    if matches!(timestamps.len(), 0) {
56        return None;
57    }
58    let mut best = 0usize;
59    let mut index = 1usize;
60    #[invariant(best@ < index@)]
61    #[invariant(index@ <= [email protected]())]
62    #[invariant(forall<i: Int> 0 <= i && i < index@ ==>
63        timestamps@[i]@ <= timestamps@[best@]@)]
64    #[invariant(forall<i: Int> 0 <= i && i < best@ ==>
65        timestamps@[i]@ < timestamps@[best@]@)]
66    #[variant([email protected]() - index@)]
67    while index < timestamps.len() {
68        if let (Some(&candidate), Some(&current)) = (timestamps.get(index), timestamps.get(best))
69            && candidate > current
70        {
71            best = index;
72        }
73        index += 1;
74    }
75    Some(best)
76}
77
78/// Compute one record's absolute offset and timestamp without signed overflow.
79#[ensures(match result {
80    Some((offset, timestamp)) => offset@ == batch_base@ + offset_delta@
81        && timestamp@ == batch_timestamp@ + timestamp_delta@,
82    None => batch_base@ + offset_delta@ < i64::MIN@
83        || batch_base@ + offset_delta@ > i64::MAX@
84        || batch_timestamp@ + timestamp_delta@ < i64::MIN@
85        || batch_timestamp@ + timestamp_delta@ > i64::MAX@,
86})]
87#[must_use]
88pub fn timestamp_record_coordinates(
89    batch_base: i64,
90    offset_delta: i32,
91    batch_timestamp: i64,
92    timestamp_delta: i64,
93) -> Option<(i64, i64)> {
94    Some((
95        batch_base.checked_add(i64::from(offset_delta))?,
96        batch_timestamp.checked_add(timestamp_delta)?,
97    ))
98}
99
100/// Advance strictly past a decoded batch, rejecting malformed or stale bounds.
101#[ensures(match result {
102    Some(next) => next@ > cursor@
103        && next@ == batch_base@ + last_offset_delta@ + 1,
104    None => last_offset_delta@ < 0
105        || batch_base@ + last_offset_delta@ + 1 > i64::MAX@
106        || batch_base@ + last_offset_delta@ < cursor@,
107})]
108#[must_use]
109pub fn timestamp_scan_next(cursor: i64, batch_base: i64, last_offset_delta: i32) -> Option<i64> {
110    if last_offset_delta < 0 {
111        return None;
112    }
113    let last = batch_base.checked_add(i64::from(last_offset_delta))?;
114    if last < cursor {
115        return None;
116    }
117    last.checked_add(1)
118}
119
120/// Double a nonzero scan window with checked arithmetic.
121#[ensures(match result {
122    Some(next) => next@ == window@ * 2 && next@ > window@,
123    None => window@ == 0 || window@ * 2 > u32::MAX@,
124})]
125#[must_use]
126pub fn timestamp_scan_window(window: u32) -> Option<u32> {
127    if window == 0 {
128        return None;
129    }
130    window.checked_mul(2)
131}
132
133/// First matching decoded record at or above the logical retention floor.
134/// A sparse byte read may start inside a batch that still contains discarded
135/// records; those records cannot satisfy the lookup.
136#[ensures(match result {
137    Some(index) => index@ < [email protected]()
138        && records@[index@].0@ >= minimum@ && records@[index@].1@ >= target@
139        && forall<i: Int> 0 <= i && i < index@
140            ==> records@[i].0@ < minimum@ || records@[i].1@ < target@,
141    None => forall<i: Int> 0 <= i && i < [email protected]()
142        ==> records@[i].0@ < minimum@ || records@[i].1@ < target@,
143})]
144#[must_use]
145pub fn first_timestamp_record_index(
146    records: &[(i64, i64)],
147    minimum: i64,
148    target: i64,
149) -> Option<usize> {
150    let mut index = 0usize;
151    #[invariant(index@ <= [email protected]())]
152    #[invariant(forall<i: Int> 0 <= i && i < index@
153        ==> records@[i].0@ < minimum@ || records@[i].1@ < target@)]
154    #[variant([email protected]() - index@)]
155    while index < records.len() {
156        if matches!(
157            records.get(index),
158            Some(&(offset, timestamp)) if offset >= minimum && timestamp >= target
159        ) {
160            return Some(index);
161        }
162        index += 1;
163    }
164    None
165}
166
167/// Decode record time according to the batch timestamp type. Append time
168/// replaces producer fields entirely, including producer arithmetic overflow.
169#[ensures(match result {
170    Some(timestamp) => timestamp@ == match append_time {
171        Some(stamp) => stamp@,
172        None => batch_timestamp@ + timestamp_delta@,
173    },
174    None => append_time == None && (batch_timestamp@ + timestamp_delta@ < i64::MIN@
175        || batch_timestamp@ + timestamp_delta@ > i64::MAX@),
176})]
177#[must_use]
178pub fn timestamp_record_time(
179    batch_timestamp: i64,
180    timestamp_delta: i64,
181    append_time: Option<i64>,
182) -> Option<i64> {
183    if let Some(stamp) = append_time {
184        return Some(stamp);
185    }
186    batch_timestamp.checked_add(timestamp_delta)
187}
188
189#[cfg(test)]
190mod tests {
191    use super::{
192        earliest_max_timestamp_index, first_timestamp_index, timestamp_record_coordinates,
193        timestamp_scan_next, timestamp_scan_window,
194    };
195
196    #[test]
197    fn selection_progress_and_overflow_are_fail_closed() {
198        let timestamps = [5, 9, 9, 3];
199        assert2::assert!(first_timestamp_index(&timestamps, 9) == Some(1));
200        assert2::assert!(first_timestamp_index(&timestamps, 10).is_none());
201        assert2::assert!(earliest_max_timestamp_index(&timestamps) == Some(1));
202        assert2::assert!(earliest_max_timestamp_index(&[]).is_none());
203        assert2::assert!(timestamp_record_coordinates(10, 2, 100, 3) == Some((12, 103)));
204        assert2::assert!(timestamp_record_coordinates(i64::MAX, 1, 0, 0).is_none());
205        assert2::assert!(timestamp_record_coordinates(0, 0, i64::MAX, 1).is_none());
206        assert2::assert!(timestamp_scan_next(10, 8, 2) == Some(11));
207        assert2::assert!(timestamp_scan_next(10, 10, 0) == Some(11));
208        assert2::assert!(timestamp_scan_next(10, 8, 1).is_none());
209        assert2::assert!(timestamp_scan_next(i64::MAX, i64::MAX, 0).is_none());
210        assert2::assert!(timestamp_scan_window(1) == Some(2));
211        assert2::assert!(timestamp_scan_window(u32::MAX).is_none());
212    }
213}
krabka_verified/timestamp.rs

krabka_verified/
timestamp.rs
