krabka_verified/storage/
future_log_swap_admission.rs

1use {super::LocalTruncationPlan, creusot_std::prelude::*};
2
3/// Admit one batch at the expected logical frontier or beyond it, and compute
4/// its inclusive last offset and exclusive successor without signed overflow.
5///
6/// A base past the frontier is a hole in the offsets, which a compacted log has
7/// and a follower of one replicates; a base below it is a duplicate.
8#[ensures(match result {
9    Some((last, next)) => expected_base@ >= 0
10        && supplied_base@ >= expected_base@
11        && last_offset_delta@ >= 0
12        && last@ == supplied_base@ + last_offset_delta@
13        && next@ == last@ + 1
14        && supplied_base@ <= last@
15        && last@ < next@,
16    None => expected_base@ < 0
17        || supplied_base@ < expected_base@
18        || last_offset_delta@ < 0
19        || supplied_base@ + last_offset_delta@ > i64::MAX@
20        || supplied_base@ + last_offset_delta@ + 1 > i64::MAX@,
21})]
22#[must_use]
23pub fn local_append_coordinates(
24    expected_base: i64,
25    supplied_base: i64,
26    last_offset_delta: i32,
27) -> Option<(i64, i64)> {
28    if expected_base < 0 || supplied_base < expected_base || last_offset_delta < 0 {
29        return None;
30    }
31    let last = supplied_base.checked_add(i64::from(last_offset_delta))?;
32    let next = last.checked_add(1)?;
33    Some((last, next))
34}
35
36/// Keep exactly the sealed-segment prefix whose bases precede the cut and keep
37/// the current active segment exactly when its base also precedes the cut.
38#[requires(forall<i: Int, j: Int>
39    0 <= i && i < j && j < [email protected]() ==> sealed_bases@[i] < sealed_bases@[j])]
40#[ensures(result.retained_sealed@ <= [email protected]())]
41#[ensures(forall<i: Int> 0 <= i && i < result.retained_sealed@ ==>
42    sealed_bases@[i]@ < cut@)]
43#[ensures(forall<i: Int> result.retained_sealed@ <= i && i < [email protected]() ==>
44    sealed_bases@[i]@ >= cut@)]
45#[ensures(result.keep_active == match active_base {
46    Some(base) => base@ < cut@,
47    None => false,
48})]
49#[must_use]
50pub fn local_truncation_plan(
51    sealed_bases: &[i64],
52    active_base: Option<i64>,
53    cut: i64,
54) -> LocalTruncationPlan {
55    let mut retained_sealed = 0usize;
56    #[invariant(retained_sealed@ <= [email protected]())]
57    #[invariant(forall<i: Int> 0 <= i && i < retained_sealed@ ==>
58        sealed_bases@[i]@ < cut@)]
59    #[variant([email protected]() - retained_sealed@)]
60    while retained_sealed < sealed_bases.len()
61        && matches!(sealed_bases.get(retained_sealed), Some(&base) if base < cut)
62    {
63        retained_sealed += 1;
64    }
65    LocalTruncationPlan {
66        retained_sealed,
67        keep_active: matches!(active_base, Some(base) if base < cut),
68    }
69}
70
71/// Convert an absolute cut to a segment-relative offset without signed or
72/// `u32` overflow. Segment bases and log cuts are nonnegative Kafka offsets.
73#[ensures(match result {
74    Some(relative) => segment_base@ >= 0
75        && cut@ >= segment_base@
76        && relative@ == cut@ - segment_base@
77        && relative@ <= u32::MAX@,
78    None => segment_base@ < 0
79        || cut@ < segment_base@
80        || cut@ - segment_base@ > u32::MAX@,
81})]
82#[must_use]
83pub fn truncation_relative_offset(segment_base: i64, cut: i64) -> Option<u32> {
84    if segment_base < 0 || cut < segment_base {
85        return None;
86    }
87    let relative = cut.abs_diff(segment_base);
88    if relative > u64::from(u32::MAX) {
89        None
90    } else {
91        // The clamp is the identity here; it hands the cast a range the
92        // compiler can see.
93        Some(relative.min(0xffff_ffff) as u32)
94    }
95}
96
97/// One decoded batch belongs to the exact retained prefix iff its inclusive
98/// last offset is below the exclusive cut.
99#[ensures(result == (batch_last@ < cut@))]
100#[must_use]
101pub const fn truncation_batch_retained(batch_last: i64, cut: i64) -> bool { batch_last < cut }
102
103/// Clamp a dependent frontier to the new log end.
104#[ensures(result@ <= frontier@)]
105#[ensures(result@ <= new_end@)]
106#[ensures(result@ == frontier@ || result@ == new_end@)]
107#[must_use]
108pub const fn truncation_frontier(frontier: i64, new_end: i64) -> i64 {
109    if frontier < new_end {
110        frontier
111    } else {
112        new_end
113    }
114}
115
116/// A future log may replace the current log only at the exact same frontier.
117#[ensures(result == (current_leo@ == future_leo@))]
118#[must_use]
119pub const fn future_log_swap_admission(current_leo: i64, future_leo: i64) -> bool {
120    current_leo == future_leo
121}
krabka_verified/storage/future_log_swap_admission.rs

krabka_verified/storage/
future_log_swap_admission.rs
