krabka_verified/
delivery.rs

1//! Scheduled-delivery visibility kernels.
2
3#[cfg(creusot)]
4use creusot_std::prelude::{Int, logic};
5use creusot_std::prelude::{ensures, requires};
6
7/// Decide whether a record batch may be exposed at the supplied clock reading.
8///
9/// A negative uncertainty is invalid and fails closed for scheduled delivery.
10/// A deadline that cannot be represented also fails closed instead of becoming
11/// visible early through saturating arithmetic.
12#[ensures(result == (!scheduled || (uncertainty_ms@ >= 0
13    && activation_ms@ + uncertainty_ms@ <= i64::MAX@
14    && activation_ms@ + uncertainty_ms@ <= now_ms@)))]
15#[must_use]
16pub fn scheduled_delivery_visible(
17    scheduled: bool,
18    uncertainty_ms: i64,
19    activation_ms: i64,
20    now_ms: i64,
21) -> bool {
22    if !scheduled {
23        return true;
24    }
25    if uncertainty_ms < 0 {
26        return false;
27    }
28    let Some(deadline_ms) = activation_ms.checked_add(uncertainty_ms) else {
29        return false;
30    };
31    deadline_ms <= now_ms
32}
33
34// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
35#[cfg(creusot)]
36#[cfg_attr(test, mutants::skip)]
37#[logic]
38fn clamp_to(value: Int, low: Int, high: Int) -> Int {
39    pearlite! { if value < low { low } else if value > high { high } else { value } }
40}
41
42/// Bound one scheduled-delivery watermark step to the live log range.
43///
44/// The current watermark is first clamped into `[log_start, log_end]` after
45/// retention or truncation. The step then moves it to the candidate when the
46/// candidate is ahead, capped at the log end:
47/// `min(max(candidate, clamped current), log_end)`. It never moves the clamped
48/// watermark backwards or past the log end.
49#[requires(log_start@ <= log_end@)]
50#[ensures(result@ == [email protected](clamp_to(current@, log_start@, log_end@)).min(log_end@))]
51#[ensures(log_start@ <= result@ && result@ <= log_end@)]
52#[ensures(current@ <= candidate@ && log_start@ <= candidate@ && candidate@ <= log_end@
53    ==> result@ == candidate@)]
54#[must_use]
55pub fn delivery_watermark_advance(
56    log_start: i64,
57    current: i64,
58    candidate: i64,
59    log_end: i64,
60) -> i64 {
61    let bounded_current = if current < log_start {
62        log_start
63    } else if current > log_end {
64        log_end
65    } else {
66        current
67    };
68    if candidate <= bounded_current {
69        bounded_current
70    } else if candidate >= log_end {
71        log_end
72    } else {
73        candidate
74    }
75}
76
77/// Decide whether the next inclusive range touches the preceding range and
78/// return the maximum end offset for a merged range.
79///
80/// The comparison is expressed without overflowing `last_high + 1` at
81/// `i64::MAX`.
82#[ensures(result.0 == (low@ <= last_high@ + 1))]
83#[ensures(result.1@ >= last_high@ && result.1@ >= high@)]
84#[ensures(result.1@ == last_high@ || result.1@ == high@)]
85#[must_use]
86pub fn coalesce_delivery_range(last_high: i64, low: i64, high: i64) -> (bool, i64) {
87    let merge = if low <= last_high {
88        true
89    } else {
90        // `low > last_high` proves `last_high < i64::MAX`.
91        low == last_high + 1
92    };
93    let merged_high = if high > last_high { high } else { last_high };
94    (merge, merged_high)
95}
96
97#[cfg(test)]
98mod tests {
99    use {
100        super::{coalesce_delivery_range, delivery_watermark_advance, scheduled_delivery_visible},
101        assert2::assert,
102    };
103
104    #[test]
105    fn visibility_never_ignores_deadline_or_overflow() {
106        assert!(scheduled_delivery_visible(false, 250, i64::MAX, i64::MIN));
107        assert!(!scheduled_delivery_visible(true, -1, 10_000, 10_000));
108        assert!(scheduled_delivery_visible(true, 0, 10_000, 10_000));
109        assert!(!scheduled_delivery_visible(true, 250, 10_000, 10_249));
110        assert!(scheduled_delivery_visible(true, 250, 10_000, 10_250));
111        assert!(!scheduled_delivery_visible(true, 1, i64::MAX, i64::MAX));
112    }
113
114    #[test]
115    fn watermark_step_is_monotonic_inside_live_bounds() {
116        assert!(delivery_watermark_advance(5, 7, 9, 10) == 9);
117        assert!(delivery_watermark_advance(5, 7, 6, 10) == 7);
118        assert!(delivery_watermark_advance(5, 7, 12, 10) == 10);
119        assert!(delivery_watermark_advance(5, 2, 8, 10) == 8);
120        assert!(delivery_watermark_advance(5, 2, 4, 10) == 5);
121        assert!(delivery_watermark_advance(5, 12, 12, 10) == 10);
122    }
123
124    #[test]
125    fn range_step_merges_only_overlap_or_adjacency() {
126        assert!(coalesce_delivery_range(4, 5, 9) == (true, 9));
127        assert!(coalesce_delivery_range(9, 6, 7) == (true, 9));
128        assert!(coalesce_delivery_range(9, 11, 12) == (false, 12));
129        assert!(coalesce_delivery_range(i64::MAX, i64::MAX, i64::MAX) == (true, i64::MAX));
130    }
131}
krabka_verified/delivery.rs

krabka_verified/
delivery.rs
