krabka_verified/broker/
coordinator_partition.rs

1use {super::FindCoordinatorAdmission, creusot_std::prelude::*};
2
3/// Decide whether one `FindCoordinator` key may proceed to coordinator lookup.
4///
5/// Kafka key types are GROUP=0, TRANSACTION=1, and SHARE=2. SHARE was added in
6/// API version 6, uses `ClusterAction` authorization, and carries a composite key
7/// that the host validates before calling this kernel. `share_key_valid` is
8/// ignored for the two non-SHARE key types. Unknown key types never inherit an
9/// allow result.
10#[ensures(key_type@ == 0 ==> result == if acl_allowed {
11    FindCoordinatorAdmission::AllowGroup
12} else {
13    FindCoordinatorAdmission::DenyGroup
14})]
15#[ensures(key_type@ == 1 ==> result == if acl_allowed {
16    FindCoordinatorAdmission::AllowTransaction
17} else {
18    FindCoordinatorAdmission::DenyTransaction
19})]
20#[ensures(key_type@ == 2 ==> result == if api_version@ < 6 || !share_key_valid {
21    FindCoordinatorAdmission::InvalidRequest
22} else if acl_allowed {
23    FindCoordinatorAdmission::AllowShare
24} else {
25    FindCoordinatorAdmission::DenyCluster
26})]
27#[ensures(key_type@ < 0 || key_type@ > 2
28    ==> result == FindCoordinatorAdmission::InvalidRequest)]
29#[must_use]
30pub fn find_coordinator_admission(
31    api_version: i16,
32    key_type: i8,
33    acl_allowed: bool,
34    share_key_valid: bool,
35) -> FindCoordinatorAdmission {
36    match key_type {
37        0 if acl_allowed => FindCoordinatorAdmission::AllowGroup,
38        0 => FindCoordinatorAdmission::DenyGroup,
39        1 if acl_allowed => FindCoordinatorAdmission::AllowTransaction,
40        1 => FindCoordinatorAdmission::DenyTransaction,
41        2 if api_version < 6 || !share_key_valid => FindCoordinatorAdmission::InvalidRequest,
42        2 if acl_allowed => FindCoordinatorAdmission::AllowShare,
43        2 => FindCoordinatorAdmission::DenyCluster,
44        _ => FindCoordinatorAdmission::InvalidRequest,
45    }
46}
47
48/// Admit an unclean-election commit only against the exact partition snapshot
49/// used to select its winner.
50#[ensures(result == (selected_partition_epoch@ == current_partition_epoch@
51    && !current_leader_alive
52    && selected_replicas@ == current_replicas@
53    && (exists<i: Int> 0 <= i && i < [email protected]()
54        && current_replicas@[i] == winner)))]
55#[must_use]
56pub fn unclean_recovery_commit_admission(
57    selected_partition_epoch: i32,
58    current_partition_epoch: i32,
59    selected_replicas: &[u64],
60    current_replicas: &[u64],
61    winner: u64,
62    current_leader_alive: bool,
63) -> bool {
64    if selected_partition_epoch != current_partition_epoch
65        || current_leader_alive
66        || selected_replicas.len() != current_replicas.len()
67    {
68        return false;
69    }
70
71    let mut winner_assigned = false;
72    let mut i = 0usize;
73    #[cfg_attr(creusot, invariant(i@ <= [email protected]()))]
74    #[cfg_attr(creusot, invariant([email protected]() == [email protected]()))]
75    #[cfg_attr(creusot, invariant(forall<k: Int> 0 <= k && k < i@
76        ==> selected_replicas@[k] == current_replicas@[k]))]
77    #[cfg_attr(creusot, invariant(winner_assigned == (exists<k: Int>
78        0 <= k && k < i@ && current_replicas@[k] == winner)))]
79    #[cfg_attr(creusot, variant([email protected]() - i@))]
80    while i < selected_replicas.len() {
81        let (Some(&selected), Some(&current)) = (selected_replicas.get(i), current_replicas.get(i))
82        else {
83            return false;
84        };
85        if selected != current {
86            return false;
87        }
88        if current == winner {
89            winner_assigned = true;
90        }
91        i += 1;
92    }
93    winner_assigned
94}
95
96/// Java `String.hashCode` over the first `limit` UTF-16 code units.
97// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
98#[cfg(creusot)]
99#[cfg_attr(test, mutants::skip)]
100#[logic]
101#[requires(0 <= limit && limit <= units.len())]
102#[variant(limit)]
103pub fn java_string_hash_prefix_model(units: Seq<u16>, limit: Int) -> i32 {
104    pearlite! {
105        if limit <= 0 {
106            0i32
107        } else {
108            java_string_hash_prefix_model(units, limit - 1) * 31i32
109                + units[limit - 1] as i32
110        }
111    }
112}
113
114// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
115#[cfg(creusot)]
116#[cfg_attr(test, mutants::skip)]
117#[logic]
118fn java_string_abs_model(hash: i32) -> i32 {
119    pearlite! {
120        if hash == i32::MIN {
121            0i32
122        } else if hash < 0i32 {
123            -hash
124        } else {
125            hash
126        }
127    }
128}
129
130// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
131#[cfg(creusot)]
132#[cfg_attr(test, mutants::skip)]
133#[logic]
134#[requires(partition_count@ > 0)]
135fn java_string_hash_partition_model(units: Seq<u16>, partition_count: i32) -> Int {
136    pearlite! {
137        java_string_abs_model(java_string_hash_prefix_model(units, units.len()))@
138            % partition_count@
139    }
140}
141
142/// Java `String.hashCode` over UTF-16 code units, followed by Kafka's
143/// `Utils.abs(hash) % partition_count` coordinator selection.
144///
145/// The host supplies `str::encode_utf16()` output so non-ASCII group ids use
146/// the same surrogate-pair semantics as the JVM. Java's `Integer.MIN_VALUE`
147/// absolute-value corner maps to zero, matching `Utils.abs`.
148#[cfg_attr(
149    creusot,
150    ensures(partition_count@ > 0 ==>
151        exists<partition: i32> result == Some(partition)
152            && partition@ == java_string_hash_partition_model(units@, partition_count))
153)]
154#[ensures(result == None ==> partition_count@ <= 0)]
155#[ensures(partition_count@ <= 0 ==> result == None)]
156#[ensures(forall<partition: i32> result == Some(partition) ==>
157    0 <= partition@ && partition@ < partition_count@)]
158#[must_use]
159pub fn java_string_hash_partition(units: &[u16], partition_count: i32) -> Option<i32> {
160    if partition_count <= 0 {
161        return None;
162    }
163
164    let mut hash = 0_i32;
165    let mut index = 0_usize;
166    #[invariant(index@ <= [email protected]())]
167    #[cfg_attr(creusot, invariant(hash == java_string_hash_prefix_model(units@, index@)))]
168    #[variant([email protected]() - index@)]
169    while index < units.len() {
170        let Some(&unit) = units.get(index) else {
171            break;
172        };
173        hash = hash.wrapping_mul(31).wrapping_add(i32::from(unit));
174        index += 1;
175    }
176    let positive = match hash {
177        i32::MIN => 0,
178        negative if negative < 0 => -negative,
179        nonnegative => nonnegative,
180    };
181    Some(positive % partition_count)
182}
krabka_verified/broker/coordinator_partition.rs

krabka_verified/broker/
coordinator_partition.rs
