krabka_verified/
produce.rs

1//! Produce batch-header admission.
2
3use creusot_std::prelude::*;
4
5#[krabka_macros::model_type]
6pub enum ProduceBatchAdmission {
7    Admit,
8    InvalidRecord,
9    InvalidTimestamp,
10}
11
12/// Compute the exclusive durability frontier for an acknowledged Produce
13/// batch. Both a fresh append and an idempotent duplicate wait for this value.
14#[ensures(match result {
15    Some(frontier) => base_offset@ >= 0
16        && last_offset_delta@ >= 0
17        && frontier@ == base_offset@ + last_offset_delta@ + 1
18        && base_offset@ < frontier@,
19    None => base_offset@ < 0
20        || last_offset_delta@ < 0
21        || base_offset@ + last_offset_delta@ + 1 > i64::MAX@,
22})]
23#[must_use]
24pub fn produce_durability_frontier(base_offset: i64, last_offset_delta: i32) -> Option<i64> {
25    if base_offset < 0 || last_offset_delta < 0 {
26        return None;
27    }
28    base_offset
29        .checked_add(i64::from(last_offset_delta))?
30        .checked_add(1)
31}
32
33#[ensures((result == ProduceBatchAdmission::InvalidRecord) ==
34    (last_offset_delta@ < 0
35        || last_offset_delta@ >= i32::MAX@
36        || records_count@ != last_offset_delta@ + 1
37        || records_count@ <= 0
38        || control_batch
39        || (producer_id@ >= 0 && base_sequence@ < 0)))]
40#[ensures((result == ProduceBatchAdmission::InvalidTimestamp) ==
41    (record_header_valid(last_offset_delta@, records_count@, control_batch, producer_id@, base_sequence@)
42        && !create_time))]
43#[ensures((result == ProduceBatchAdmission::Admit) ==
44    (record_header_valid(last_offset_delta@, records_count@, control_batch, producer_id@, base_sequence@)
45        && create_time))]
46#[must_use]
47pub fn produce_batch_admission(
48    last_offset_delta: i32,
49    records_count: i32,
50    control_batch: bool,
51    producer_id: i64,
52    base_sequence: i32,
53    create_time: bool,
54) -> ProduceBatchAdmission {
55    if last_offset_delta < 0
56        || last_offset_delta == i32::MAX
57        || last_offset_delta + 1 != records_count
58        || records_count <= 0
59        || control_batch
60        || (producer_id >= 0 && base_sequence < 0)
61    {
62        ProduceBatchAdmission::InvalidRecord
63    } else if !create_time {
64        ProduceBatchAdmission::InvalidTimestamp
65    } else {
66        ProduceBatchAdmission::Admit
67    }
68}
69
70// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
71#[cfg(creusot)]
72#[cfg_attr(test, mutants::skip)]
73#[logic(open)]
74/// A data batch has a representable successor, exact positive cardinality and a valid producer sequence.
75fn record_header_valid(
76    delta: Int,
77    count: Int,
78    control: bool,
79    producer: Int,
80    sequence: Int,
81) -> bool {
82    pearlite! { 0 <= delta && delta < i32::MAX@ && count == delta + 1 && count > 0
83    && !control && (producer < 0 || sequence >= 0) }
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89
90    #[test]
91    fn durability_frontier_is_exact_and_fail_closed() {
92        assert2::assert!(produce_durability_frontier(0, 0) == Some(1));
93        assert2::assert!(produce_durability_frontier(10, 2) == Some(13));
94        assert2::assert!(produce_durability_frontier(-1, 0).is_none());
95        assert2::assert!(produce_durability_frontier(10, -1).is_none());
96        assert2::assert!(produce_durability_frontier(i64::MAX, 0).is_none());
97        assert2::assert!(produce_durability_frontier(i64::MAX - 1, 0) == Some(i64::MAX));
98    }
99
100    #[test]
101    fn produce_header_admission_is_ordered_and_exhaustive() {
102        use ProduceBatchAdmission::{Admit, InvalidRecord, InvalidTimestamp};
103
104        assert2::assert!(produce_batch_admission(0, 1, false, -1, -1, true) == Admit);
105        assert2::assert!(produce_batch_admission(1, 2, false, 7, 0, true) == Admit);
106        assert2::assert!(produce_batch_admission(0, 2, false, -1, -1, true) == InvalidRecord);
107        assert2::assert!(produce_batch_admission(-1, 0, false, -1, -1, true) == InvalidRecord);
108        assert2::assert!(
109            produce_batch_admission(i32::MAX, i32::MAX, false, -1, -1, true) == InvalidRecord
110        );
111        assert2::assert!(produce_batch_admission(0, 0, false, -1, -1, true) == InvalidRecord);
112        assert2::assert!(produce_batch_admission(0, 1, true, -1, -1, true) == InvalidRecord);
113        assert2::assert!(produce_batch_admission(0, 1, false, 7, -1, true) == InvalidRecord);
114        assert2::assert!(produce_batch_admission(0, 1, false, -1, -1, false) == InvalidTimestamp);
115    }
116}
krabka_verified/produce.rs

krabka_verified/
produce.rs
