krabka_verified/raft/
metadata_record_offset_deltas.rs

1use super::{FetchContent, FetchFence, FetchResponseFacts, FetchResponseMutation};
2#[cfg(creusot)]
3use creusot_std::prelude::*;
4
5// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
6#[cfg(creusot)]
7#[cfg_attr(test, mutants::skip)]
8#[logic(open)]
9/// The response is a successful answer from the leader this node already
10/// follows, in this node's epoch: only such a response may change the log or
11/// the HWM (`KafkaRaftClient.handleFetchResponse` applies content only for
12/// `Errors.NONE` from the followed leader).
13pub fn leader_fence_holds(fence: FetchFence, response: FetchResponseFacts) -> bool {
14    pearlite! {
15        fence.role_leader == Some(response.from)
16            && fence.current_leader == Some(response.from)
17            && response.leader == Some(response.from)
18            && response.epoch == fence.current_epoch
19            && response.error_none
20    }
21}
22
23// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
24#[cfg(creusot)]
25#[cfg_attr(test, mutants::skip)]
26#[logic(open)]
27/// Kafka's `maybeHandleCommonResponse` discovery case: a response in this
28/// node's epoch names a leader while the node knows none. It applies whatever
29/// the error is, since a follower that knows the leader answers
30/// `NOT_LEADER_OR_FOLLOWER` naming it. A newer epoch is the host's
31/// `BeginQuorumEpoch` path and never reaches this kernel.
32pub fn discovery_applies(fence: FetchFence, response: FetchResponseFacts) -> bool {
33    pearlite! {
34        fence.discovering
35            && fence.role_leader == None
36            && fence.current_leader == None
37            && response.epoch == fence.current_epoch
38            && response.leader != None
39    }
40}
41
42/// Fence a Fetch response against the live role, leader, and epoch, then
43/// select exactly one mutation path.
44///
45/// A discovery response carries no mutation: the node attaches to the
46/// advertised leader and fetches again under the leader fence, which admits
47/// content only from that leader itself. So no response is ever applied on
48/// the strength of who the responder is assumed to be.
49#[cfg_attr(creusot, ensures((result == FetchResponseMutation::Discover)
50    == discovery_applies(fence, response)))]
51#[cfg_attr(creusot, ensures((result == FetchResponseMutation::Reject)
52    == (!discovery_applies(fence, response) && !leader_fence_holds(fence, response))))]
53#[cfg_attr(creusot, ensures((result == FetchResponseMutation::Snapshot)
54    == (leader_fence_holds(fence, response) && content.has_snapshot)))]
55#[cfg_attr(creusot, ensures((result == FetchResponseMutation::Truncate)
56    == (leader_fence_holds(fence, response)
57        && !content.has_snapshot
58        && content.has_divergence)))]
59#[cfg_attr(creusot, ensures((result == FetchResponseMutation::Append)
60    == (leader_fence_holds(fence, response)
61        && !content.has_snapshot
62        && !content.has_divergence
63        && content.has_records)))]
64#[cfg_attr(creusot, ensures((result == FetchResponseMutation::HighWatermark)
65    == (leader_fence_holds(fence, response)
66        && !content.has_snapshot
67        && !content.has_divergence
68        && !content.has_records)))]
69#[must_use]
70pub fn fetch_response_mutation(
71    fence: FetchFence,
72    response: FetchResponseFacts,
73    content: FetchContent,
74) -> FetchResponseMutation {
75    if fence.discovering
76        && fence.role_leader.is_none()
77        && fence.current_leader.is_none()
78        && response.epoch == fence.current_epoch
79        && response.leader.is_some()
80    {
81        FetchResponseMutation::Discover
82    } else if fence.role_leader != Some(response.from)
83        || fence.current_leader != Some(response.from)
84        || response.leader != Some(response.from)
85        || response.epoch != fence.current_epoch
86        || !response.error_none
87    {
88        FetchResponseMutation::Reject
89    } else if content.has_snapshot {
90        FetchResponseMutation::Snapshot
91    } else if content.has_divergence {
92        FetchResponseMutation::Truncate
93    } else if content.has_records {
94        FetchResponseMutation::Append
95    } else {
96        FetchResponseMutation::HighWatermark
97    }
98}
99
100/// Advance a high watermark monotonically without passing the log end.
101#[must_use]
102#[cfg_attr(creusot, ensures(result@ >= previous@))]
103#[cfg_attr(creusot, ensures(previous@ <= log_end@ ==> result@ <= log_end@))]
104#[cfg_attr(creusot, ensures(result@ ==
105    if requested@ <= log_end@ {
106        if previous@ >= requested@ { previous@ } else { requested@ }
107    } else if previous@ >= log_end@ { previous@ } else { log_end@ }))]
108pub const fn advance_high_watermark(previous: i64, requested: i64, log_end: i64) -> i64 {
109    let clamped = if requested <= log_end {
110        requested
111    } else {
112        log_end
113    };
114    if previous >= clamped {
115        previous
116    } else {
117        clamped
118    }
119}
120
121/// Whether `value` lies in `[start, end)`.
122#[must_use]
123#[cfg_attr(creusot, ensures(result == (start@ <= value@ && value@ < end@)))]
124pub const fn in_half_open_window(value: i64, start: i64, end: i64) -> bool {
125    start <= value && value < end
126}
127
128/// Whether a monotonic frontier has reached a target offset.
129#[must_use]
130#[cfg_attr(creusot, ensures(result == (frontier@ >= target@)))]
131pub const fn frontier_reaches(frontier: i64, target: i64) -> bool { frontier >= target }
132
133/// Return the length of the strictly ordered control-record prefix below a
134/// half-open frontier.
135#[cfg_attr(creusot, requires(crate::sequence::strictly_increasing(offsets@)))]
136#[cfg_attr(creusot, ensures(result@ <= [email protected]()))]
137#[cfg_attr(creusot, ensures(forall<i: Int>
138    0 <= i && i < result@ ==> offsets@[i]@ < frontier@))]
139#[cfg_attr(creusot, ensures(forall<i: Int>
140    result@ <= i && i < [email protected]() ==> frontier@ <= offsets@[i]@))]
141#[must_use]
142pub fn control_history_frontier(offsets: &[i64], frontier: i64) -> usize {
143    let mut lo = 0usize;
144    let mut hi = offsets.len();
145    #[cfg_attr(creusot, invariant(lo@ <= hi@ && hi@ <= [email protected]()))]
146    #[cfg_attr(creusot, invariant(forall<i: Int>
147        0 <= i && i < lo@ ==> offsets@[i]@ < frontier@))]
148    #[cfg_attr(creusot, invariant(forall<i: Int>
149        hi@ <= i && i < [email protected]() ==> frontier@ <= offsets@[i]@))]
150    #[cfg_attr(creusot, variant(hi - lo))]
151    while lo < hi {
152        let mid = lo + (hi - lo) / 2;
153        if matches!(offsets.get(mid), Some(&offset) if offset < frontier) {
154            lo = mid + 1;
155        } else {
156            hi = mid;
157        }
158    }
159    lo
160}
161
162/// The contiguous offset deltas `0, 1, ..., record_count - 1` of one metadata
163/// batch, in record order; the last element is the batch's `lastOffsetDelta`.
164///
165/// An empty batch, and a batch whose final delta Kafka's signed 32-bit
166/// `lastOffsetDelta` cannot represent, fail closed. The deltas are counted in
167/// `i32` beside the vector's length rather than narrowed from a `usize`
168/// index, so the one executable body is the one Creusot proves.
169#[must_use]
170#[cfg_attr(creusot, ensures(match result {
171    Some(deltas) => record_count@ > 0
172        && record_count@ <= i32::MAX@ + 1
173        && [email protected]() == record_count@
174        && forall<i: Int> 0 <= i && i < [email protected]() ==> deltas@[i]@ == i,
175    None => record_count@ == 0 || record_count@ > i32::MAX@ + 1,
176}))]
177pub fn metadata_record_offset_deltas(record_count: usize) -> Option<Vec<i32>> {
178    // `i32::MAX + 1` records have final delta `i32::MAX`.
179    let max_records: usize = 0x8000_0000;
180    if record_count == 0 || record_count > max_records {
181        return None;
182    }
183    let mut deltas = Vec::with_capacity(record_count);
184    deltas.push(0);
185    let mut last: i32 = 0;
186    #[cfg_attr(creusot, invariant(1 <= [email protected]() && [email protected]() <= record_count@))]
187    #[cfg_attr(creusot, invariant(last@ == [email protected]() - 1))]
188    #[cfg_attr(creusot, invariant(forall<i: Int>
189        0 <= i && i < [email protected]() ==> deltas@[i]@ == i))]
190    #[cfg_attr(creusot, variant(record_count@ - [email protected]()))]
191    while deltas.len() < record_count {
192        // `last + 1 == deltas.len() < record_count <= i32::MAX + 1`.
193        last += 1;
194        deltas.push(last);
195    }
196    Some(deltas)
197}
krabka_verified/raft/metadata_record_offset_deltas.rs

krabka_verified/raft/
metadata_record_offset_deltas.rs
