krabka_verified/remote_read/
relative_offset.rs

1use creusot_std::prelude::*;
2
3/// Compute the inclusive end of a capped remote-segment fetch.
4///
5/// A zero cap means read to the segment end. A finite end exists only when the
6/// exclusive mathematical end stays strictly inside the segment.
7#[ensures(match result {
8    Some(end) => max_bytes@ > 0
9        && start_position@ + max_bytes@ < segment_size@
10        && end@ == start_position@ + max_bytes@ - 1
11        && end@ < segment_size@,
12    None => max_bytes@ == 0 || start_position@ + max_bytes@ >= segment_size@,
13})]
14#[must_use]
15pub fn remote_fetch_end_position(
16    start_position: u32,
17    segment_size: u32,
18    max_bytes: u32,
19) -> Option<u32> {
20    if max_bytes == 0 {
21        return None;
22    }
23    // An end past `u32::MAX` is past every `u32` segment size too.
24    let exclusive_end = start_position.checked_add(max_bytes)?;
25    if exclusive_end >= segment_size {
26        None
27    } else {
28        Some(exclusive_end - 1)
29    }
30}
31
32/// Admit a remote segment for one requested offset and derive its relative
33/// offset for Kafka's `u32` sparse index.
34///
35/// A segment is usable only after its copy finished, when it contains the
36/// requested offset, and when the offset falls inside the requested leader
37/// epoch's `[epoch_start, next_epoch_start)` subrange. The last epoch has no
38/// next boundary and runs through the segment end. A segment wider than the
39/// relative-index representation fails closed rather than truncating or
40/// defaulting the relative offset.
41#[ensures(match result {
42    Some(delta) => match epoch_start {
43        Some(epoch_start) => copy_finished
44            && start_offset@ <= epoch_start@
45            && epoch_start@ <= requested_offset@
46            && requested_offset@ <= end_offset@
47            && match next_epoch_start {
48                Some(next) => epoch_start@ < next@
49                    && next@ <= end_offset@
50                    && requested_offset@ < next@,
51                None => true,
52            }
53            && delta@ == requested_offset@ - start_offset@,
54        None => false,
55    },
56    None => match epoch_start {
57        Some(epoch_start) => !copy_finished
58            || epoch_start@ < start_offset@
59            || requested_offset@ < epoch_start@
60            || requested_offset@ > end_offset@
61            || match next_epoch_start {
62                Some(next) => next@ <= epoch_start@
63                    || next@ > end_offset@
64                    || requested_offset@ >= next@,
65                None => false,
66            }
67            || requested_offset@ - start_offset@ > u32::MAX@,
68        None => true,
69    },
70})]
71#[must_use]
72pub fn remote_read_relative_offset(
73    start_offset: i64,
74    end_offset: i64,
75    requested_offset: i64,
76    copy_finished: bool,
77    epoch_start: Option<i64>,
78    next_epoch_start: Option<i64>,
79) -> Option<u32> {
80    let epoch_start = epoch_start?;
81    if !copy_finished
82        || epoch_start < start_offset
83        || requested_offset < epoch_start
84        || requested_offset > end_offset
85    {
86        return None;
87    }
88    match next_epoch_start {
89        Some(next) if next <= epoch_start || next > end_offset || requested_offset >= next => {
90            return None;
91        }
92        _ => {}
93    }
94
95    // `start_offset <= epoch_start <= requested_offset`, so the distance is
96    // the delta and cannot overflow.
97    let delta = requested_offset.abs_diff(start_offset);
98    if delta > u64::from(u32::MAX) {
99        None
100    } else {
101        // The clamp is the identity here; it hands the cast a range the
102        // compiler can see.
103        Some(delta.min(0xffff_ffff) as u32)
104    }
105}
krabka_verified/remote_read/relative_offset.rs

krabka_verified/remote_read/
relative_offset.rs
