krabka_verified/
quorum_state.rs

1//! `KRaft` quorum-state persistence admission.
2
3use creusot_std::prelude::ensures;
4#[cfg(creusot)]
5use creusot_std::prelude::{DeepModel, logic};
6
7/// Whether an in-memory quorum state can be represented exactly by Kafka's
8/// signed, versioned JSON fields.
9#[krabka_macros::model_type]
10pub enum QuorumStateWriteDecision {
11    Reject,
12    Accept,
13}
14
15/// The in-memory quorum state has an exact representation in Kafka's
16/// quorum-state JSON: a supported schema version (0 or 1), and a term, leader
17/// ID, and voted ID that fit their signed `int32` fields when present. Schema
18/// v0 also lists the voter IDs, so they must fit too; schema v1 does not
19/// carry them.
20// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
21#[cfg(creusot)]
22#[cfg_attr(test, mutants::skip)]
23#[logic]
24pub fn fits_signed_json(
25    kraft_version: u16,
26    leader_epoch: u32,
27    has_leader: bool,
28    leader_id: u64,
29    has_vote: bool,
30    voted_id: u64,
31    all_voter_ids_fit: bool,
32) -> bool {
33    pearlite! {
34        kraft_version@ <= 1
35            && leader_epoch@ <= i32::MAX@
36            && (!has_leader || leader_id@ <= i32::MAX@)
37            && (!has_vote || voted_id@ <= i32::MAX@)
38            && (kraft_version@ == 1 || all_voter_ids_fit)
39    }
40}
41
42/// Reject states that would be clamped, lose an identity, or select an
43/// unsupported schema when persisted. The one ensures pins both variants:
44/// `Accept` exactly when `fits_signed_json` holds.
45#[ensures((result == QuorumStateWriteDecision::Accept) == fits_signed_json(
46    kraft_version,
47    leader_epoch,
48    has_leader,
49    leader_id,
50    has_vote,
51    voted_id,
52    all_voter_ids_fit,
53))]
54#[must_use]
55pub fn quorum_state_write_decision(
56    kraft_version: u16,
57    leader_epoch: u32,
58    has_leader: bool,
59    leader_id: u64,
60    has_vote: bool,
61    voted_id: u64,
62    all_voter_ids_fit: bool,
63) -> QuorumStateWriteDecision {
64    if kraft_version > 1
65        || leader_epoch > 2_147_483_647
66        || (has_leader && leader_id > 2_147_483_647)
67        || (has_vote && voted_id > 2_147_483_647)
68        || (kraft_version == 0 && !all_voter_ids_fit)
69    {
70        QuorumStateWriteDecision::Reject
71    } else {
72        QuorumStateWriteDecision::Accept
73    }
74}
75
76/// How a parsed quorum-state record may restore the durable vote.
77#[krabka_macros::model_type]
78pub enum QuorumStateLoadDecision {
79    Reject,
80    RestoreNoVote,
81    RestoreVote,
82}
83
84/// Admit only schema v0 or v1 records with a nonnegative term, the exact vote
85/// sentinel, and valid version-specific fields.
86///
87/// Schema v1's no-vote sentinel must carry the nil directory sentinel written
88/// by the encoder.
89#[ensures((result == QuorumStateLoadDecision::Reject) == (
90    (data_version@ != 0 && data_version@ != 1)
91        || leader_epoch@ < 0
92        || voted_id@ < -1
93        || !version_fields_valid
94        || (data_version@ == 1 && !voted_directory_valid)
95        || (data_version@ == 1 && voted_id@ == -1 && !voted_directory_nil)
96))]
97#[ensures((result == QuorumStateLoadDecision::RestoreNoVote) == (
98    (data_version@ == 0 || data_version@ == 1)
99        && leader_epoch@ >= 0
100        && voted_id@ == -1
101        && version_fields_valid
102        && (data_version@ == 0 || (voted_directory_valid && voted_directory_nil))
103))]
104#[ensures((result == QuorumStateLoadDecision::RestoreVote) == (
105    (data_version@ == 0 || data_version@ == 1)
106        && leader_epoch@ >= 0
107        && voted_id@ >= 0
108        && version_fields_valid
109        && (data_version@ == 0 || voted_directory_valid)
110))]
111#[must_use]
112pub fn quorum_state_load_decision(
113    data_version: i32,
114    leader_epoch: i32,
115    voted_id: i32,
116    version_fields_valid: bool,
117    voted_directory_valid: bool,
118    voted_directory_nil: bool,
119) -> QuorumStateLoadDecision {
120    if (data_version != 0 && data_version != 1)
121        || leader_epoch < 0
122        || voted_id < -1
123        || !version_fields_valid
124        || (data_version == 1 && !voted_directory_valid)
125        || (data_version == 1 && voted_id == -1 && !voted_directory_nil)
126    {
127        QuorumStateLoadDecision::Reject
128    } else if voted_id == -1 {
129        QuorumStateLoadDecision::RestoreNoVote
130    } else {
131        QuorumStateLoadDecision::RestoreVote
132    }
133}
134
135#[cfg(test)]
136mod tests {
137    use {
138        super::{
139            QuorumStateLoadDecision, QuorumStateWriteDecision, quorum_state_load_decision,
140            quorum_state_write_decision,
141        },
142        assert2::check,
143    };
144
145    #[yare::parameterized(
146        v0_max = {
147            0,
148            i32::MAX as u32,
149            true,
150            i32::MAX as u64,
151            true,
152            i32::MAX as u64,
153            true,
154            QuorumStateWriteDecision::Accept,
155        },
156        v1_max = {
157            1,
158            i32::MAX as u32,
159            true,
160            i32::MAX as u64,
161            true,
162            i32::MAX as u64,
163            false,
164            QuorumStateWriteDecision::Accept,
165        },
166        unsupported_version = { 2, 0, false, 0, false, 0, true, QuorumStateWriteDecision::Reject },
167        epoch_overflow = {
168            0,
169            i32::MAX as u32 + 1,
170            false,
171            0,
172            false,
173            0,
174            true,
175            QuorumStateWriteDecision::Reject,
176        },
177        leader_overflow = {
178            0,
179            0,
180            true,
181            i32::MAX as u64 + 1,
182            false,
183            0,
184            true,
185            QuorumStateWriteDecision::Reject,
186        },
187        vote_overflow = {
188            0,
189            0,
190            false,
191            0,
192            true,
193            i32::MAX as u64 + 1,
194            true,
195            QuorumStateWriteDecision::Reject,
196        },
197        ignored_absent_ids = {
198            0,
199            0,
200            false,
201            u64::MAX,
202            false,
203            u64::MAX,
204            true,
205            QuorumStateWriteDecision::Accept,
206        },
207        v0_voter_overflow = { 0, 0, false, 0, false, 0, false, QuorumStateWriteDecision::Reject },
208    )]
209    fn writes_reject_every_lossy_boundary(
210        version: u16,
211        epoch: u32,
212        has_leader: bool,
213        leader: u64,
214        has_vote: bool,
215        vote: u64,
216        voters_fit: bool,
217        expected: QuorumStateWriteDecision,
218    ) {
219        check!(
220            quorum_state_write_decision(
221                version, epoch, has_leader, leader, has_vote, vote, voters_fit
222            ) == expected
223        );
224    }
225
226    #[yare::parameterized(
227        v0_no_vote = { 0, 0, -1, true, false, false, QuorumStateLoadDecision::RestoreNoVote },
228        v0_vote = { 0, i32::MAX, 0, true, false, false, QuorumStateLoadDecision::RestoreVote },
229        v1_no_vote = { 1, 0, -1, true, true, true, QuorumStateLoadDecision::RestoreNoVote },
230        v1_vote = {
231            1,
232            i32::MAX,
233            i32::MAX,
234            true,
235            true,
236            false,
237            QuorumStateLoadDecision::RestoreVote,
238        },
239        negative_version = { -1, 0, -1, true, false, false, QuorumStateLoadDecision::Reject },
240        unsupported_version = { 2, 0, -1, true, true, true, QuorumStateLoadDecision::Reject },
241        negative_epoch = { 0, -1, -1, true, false, false, QuorumStateLoadDecision::Reject },
242        invalid_vote_sentinel = { 0, 0, -2, true, false, false, QuorumStateLoadDecision::Reject },
243        invalid_v0_fields = { 0, 0, -1, false, false, false, QuorumStateLoadDecision::Reject },
244        v1_missing_directory = { 1, 0, 1, true, false, false, QuorumStateLoadDecision::Reject },
245        v1_non_nil_no_vote = { 1, 0, -1, true, true, false, QuorumStateLoadDecision::Reject },
246    )]
247    fn loads_require_exact_sentinel_and_version_fields(
248        version: i32,
249        epoch: i32,
250        vote: i32,
251        fields: bool,
252        dir_valid: bool,
253        dir_nil: bool,
254        expected: QuorumStateLoadDecision,
255    ) {
256        check!(
257            quorum_state_load_decision(version, epoch, vote, fields, dir_valid, dir_nil)
258                == expected
259        );
260    }
261}
krabka_verified/quorum_state.rs

krabka_verified/
quorum_state.rs
