krabka_verified/producer/
sequence_decision.rs

1#[cfg(creusot)]
2use super::{ProducerDecision, ProducerEntryFacts, RetainedSequenceRange};
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)]
9pub fn sequence_modulo_2_31(sequence: Int) -> Int {
10    pearlite! { sequence.rem_euclid(2147483648) }
11}
12
13/// Advance a Kafka producer sequence modulo `2^31`.
14///
15/// The result is `sequence + increment` reduced modulo `2^31`. For the
16/// nonnegative operands Kafka uses, that is exactly
17/// `DefaultRecordBatch.incrementSequence`: the sum when it fits in an `i32`,
18/// otherwise the sum minus `2^31`.
19#[ensures(result@ == sequence_modulo_2_31(sequence@ + increment@))]
20#[must_use]
21pub fn increment_sequence(sequence: i32, increment: i32) -> i32 {
22    // Reduce the sum into `[0, 2^31)` without leaving `i32`: adding `2^31` is
23    // subtracting `i32::MIN`, and subtracting `2^31` is adding it.
24    match sequence.checked_add(increment) {
25        Some(sum) if sum >= 0 => sum,
26        Some(sum) => sum - i32::MIN,
27        // The sum passed `i32::MAX`, so both operands are positive.
28        None if increment > 0 => sequence + (increment + i32::MIN),
29        // The sum fell below `i32::MIN`, so both operands are negative.
30        None => (sequence - i32::MIN) + (increment - i32::MIN),
31    }
32}
33
34/// Move a Kafka producer sequence backwards modulo `2^31`.
35///
36/// The result is `sequence - decrement` reduced modulo `2^31`. For the
37/// nonnegative operands Kafka uses, that is exactly
38/// `DefaultRecordBatch.decrementSequence`: the difference when it is not
39/// negative, otherwise the difference plus `2^31`.
40#[ensures(result@ == sequence_modulo_2_31(sequence@ - decrement@))]
41#[must_use]
42pub fn decrement_sequence(sequence: i32, decrement: i32) -> i32 {
43    // The same reduction as `increment_sequence`, for a difference.
44    match sequence.checked_sub(decrement) {
45        Some(difference) if difference >= 0 => difference,
46        Some(difference) => difference - i32::MIN,
47        // The difference passed `i32::MAX`: `sequence >= 0 > decrement`.
48        None if decrement < 0 => sequence + (i32::MIN - decrement),
49        // The difference fell below `i32::MIN`: `sequence < 0 < decrement`.
50        None => (sequence - i32::MIN) - (decrement + i32::MIN),
51    }
52}
53
54// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
55#[cfg(creusot)]
56#[cfg_attr(test, mutants::skip)]
57#[logic(open)]
58pub fn retained_matches(
59    slot: Option<RetainedSequenceRange>,
60    base_sequence: i32,
61    last_offset_delta: i32,
62) -> bool {
63    pearlite! {
64        match slot {
65            Some(range) => range.base_sequence@ == base_sequence@
66                && range.last_sequence@
67                    == sequence_modulo_2_31(base_sequence@ + last_offset_delta@),
68            None => false,
69        }
70    }
71}
72
73// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
74#[cfg(creusot)]
75#[cfg_attr(test, mutants::skip)]
76#[logic(open)]
77pub fn retained_duplicate_exists(
78    entry: Option<ProducerEntryFacts>,
79    retained: Seq<Option<RetainedSequenceRange>>,
80    producer_epoch: i16,
81    base_sequence: i32,
82    last_offset_delta: i32,
83) -> bool {
84    pearlite! {
85        match entry {
86            Some(tracked) => tracked.epoch@ == producer_epoch@
87                && exists<i: Int> 0 <= i && i < retained.len()
88                    && retained_matches(retained[i], base_sequence, last_offset_delta),
89            None => false,
90        }
91    }
92}
93
94// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
95#[cfg(creusot)]
96#[cfg_attr(test, mutants::skip)]
97#[logic(open)]
98pub fn first_retained_duplicate(
99    entry: Option<ProducerEntryFacts>,
100    retained: Seq<Option<RetainedSequenceRange>>,
101    producer_epoch: i16,
102    base_sequence: i32,
103    last_offset_delta: i32,
104    index: Int,
105) -> bool {
106    pearlite! {
107        match entry {
108            Some(tracked) => tracked.epoch@ == producer_epoch@
109                && 0 <= index && index < retained.len()
110                && retained_matches(retained[index], base_sequence, last_offset_delta)
111                && forall<j: Int> 0 <= j && j < index
112                    ==> !retained_matches(retained[j], base_sequence, last_offset_delta),
113            None => false,
114        }
115    }
116}
117
118// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
119#[cfg(creusot)]
120#[cfg_attr(test, mutants::skip)]
121#[logic(open)]
122pub fn sequence_decision(
123    entry: Option<ProducerEntryFacts>,
124    producer_epoch: i16,
125    base_sequence: i32,
126    log_empty: bool,
127    trunk_rules: bool,
128) -> ProducerDecision {
129    pearlite! {
130        match entry {
131            None => if trunk_rules && log_empty && base_sequence@ != 0 {
132                ProducerDecision::OutOfOrder
133            } else {
134                ProducerDecision::Append
135            },
136            Some(tracked) => if producer_epoch@ < tracked.epoch@ {
137                ProducerDecision::Fenced
138            } else if producer_epoch@ > tracked.epoch@ {
139                if base_sequence@ == 0 { ProducerDecision::Append } else { ProducerDecision::OutOfOrder }
140            } else if base_sequence@ == sequence_modulo_2_31(tracked.last_sequence@ + 1) {
141                ProducerDecision::Append
142            } else {
143                ProducerDecision::OutOfOrder
144            },
145        }
146    }
147}
krabka_verified/producer/sequence_decision.rs

krabka_verified/producer/
sequence_decision.rs
