krabka_verified/reconfiguration/
voter_reconfiguration_decision.rs

1use super::{
2    CurrentVoterSet, ReconfigurationLeadership, TargetMembership, TargetVoter, VoterChangeKind,
3    VoterChangeRequest, VoterReconfigurationDecision, VoterReconfigurationPlan,
4};
5#[cfg(creusot)]
6use super::{
7    admitted_plan, is_admit, may_reconfigure, update_key_matches, voter_reconfiguration_rejection,
8    voter_set_removes,
9};
10#[cfg(creusot)]
11use creusot_std::prelude::*;
12
13#[cfg(creusot)]
14// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
15#[cfg_attr(test, mutants::skip)]
16#[logic(open)]
17fn voter_set_change_admission_valid(
18    voters: CurrentVoterSet,
19    kind: VoterChangeKind,
20    decision: VoterReconfigurationDecision,
21) -> bool {
22    pearlite! {
23        match decision {
24            VoterReconfigurationDecision::Admit(plan) =>
25                voters.kraft_version@ == 1 && admitted_plan(voters, kind, plan),
26            VoterReconfigurationDecision::UnsupportedKraftVersion => voters.kraft_version@ != 1,
27            _ => true,
28        }
29    }
30}
31
32/// Admit one fresh KIP-853 operation and construct its exact result shape.
33///
34/// The result is exactly the first rejection of
35/// `voter_reconfiguration_rejection`, which follows the precedence of Kafka's
36/// handlers, and otherwise an admission with the plan `admitted_plan`
37/// describes. Every admitted change satisfies `may_reconfigure`.
38#[cfg_attr(creusot, ensures(match voter_reconfiguration_rejection(leadership, voters, request, target) {
39    Some(rejection) => result == rejection,
40    None => match result {
41        VoterReconfigurationDecision::Admit(plan) => admitted_plan(voters, request.kind, plan),
42        _ => false,
43    },
44}))]
45#[cfg_attr(creusot, ensures(is_admit(result) ==> may_reconfigure(leadership, voters)))]
46#[cfg_attr(creusot, ensures(forall<plan: VoterReconfigurationPlan>
47    result == VoterReconfigurationDecision::Admit(plan)
48        ==> plan.next_voter_count@ > 0 && plan.next_kraft_version@ <= 1))]
49#[must_use]
50pub fn voter_reconfiguration_decision(
51    leadership: ReconfigurationLeadership,
52    voters: CurrentVoterSet,
53    request: VoterChangeRequest,
54    target: TargetVoter,
55) -> VoterReconfigurationDecision {
56    if !leadership.is_leader {
57        return VoterReconfigurationDecision::NotLeader;
58    }
59    if !leadership.no_pending_change {
60        return VoterReconfigurationDecision::InProgress;
61    }
62    if !leadership.epoch_committed {
63        return VoterReconfigurationDecision::EpochUncommitted;
64    }
65    if !voters.latest_controls_committed {
66        return VoterReconfigurationDecision::InProgress;
67    }
68    if voters.voter_count == 0 {
69        return VoterReconfigurationDecision::EmptyCurrentVoterSet;
70    }
71    if voters.kraft_version > 1 {
72        return VoterReconfigurationDecision::InvalidVersionTransition;
73    }
74    match request.kind {
75        VoterChangeKind::Add => add_decision(voters, target),
76        VoterChangeKind::Remove => remove_decision(voters, target),
77        VoterChangeKind::Update => update_decision(voters, target),
78        VoterChangeKind::FinalizeKraftVersion => finalize_decision(voters, request),
79    }
80}
81
82// Add and Remove both publish only the successor voter-set record.
83#[cfg_attr(creusot, ensures(match result {
84    VoterReconfigurationDecision::Admit(plan) =>
85        plan.next_voter_count@ == next_voter_count@
86        && plan.next_kraft_version@ == next_kraft_version@
87        && plan.write_voters && !plan.write_kraft_version && !plan.preflight_only,
88    _ => false,
89}))]
90fn admit_voter_set_change(
91    next_voter_count: usize,
92    next_kraft_version: u16,
93) -> VoterReconfigurationDecision {
94    VoterReconfigurationDecision::Admit(VoterReconfigurationPlan {
95        next_voter_count,
96        next_kraft_version,
97        write_voters: true,
98        write_kraft_version: false,
99        preflight_only: false,
100    })
101}
102
103/// `AddVoterHandler`: version, duplicate id, `ApiVersions` range, catch-up.
104#[cfg_attr(creusot, requires(decision_voter_set_valid(voters)))]
105#[cfg_attr(
106    creusot,
107    ensures(voter_set_change_admission_valid(voters, VoterChangeKind::Add, result,))
108)]
109#[cfg_attr(creusot, ensures(match result {
110    VoterReconfigurationDecision::Admit(_) =>
111        target.membership == TargetMembership::Absent
112            && target.version_compatible
113            && target.caught_up
114            && voters.voter_count@ < usize::MAX@,
115    VoterReconfigurationDecision::UnsupportedKraftVersion => true,
116    VoterReconfigurationDecision::DuplicateVoter => voters.kraft_version@ == 1
117        && target.membership != TargetMembership::Absent,
118    VoterReconfigurationDecision::IncompatibleVoter => voters.kraft_version@ == 1
119        && target.membership == TargetMembership::Absent
120        && !target.version_compatible,
121    VoterReconfigurationDecision::VoterNotCaughtUp => voters.kraft_version@ == 1
122        && target.membership == TargetMembership::Absent
123        && target.version_compatible
124        && !target.caught_up,
125    VoterReconfigurationDecision::InvalidVersionTransition => voters.kraft_version@ == 1
126        && target.membership == TargetMembership::Absent
127        && target.version_compatible
128        && target.caught_up
129        && voters.voter_count@ == usize::MAX@,
130    _ => false,
131}))]
132fn add_decision(voters: CurrentVoterSet, target: TargetVoter) -> VoterReconfigurationDecision {
133    if voters.kraft_version != 1 {
134        return VoterReconfigurationDecision::UnsupportedKraftVersion;
135    }
136    if !matches!(target.membership, TargetMembership::Absent) {
137        return VoterReconfigurationDecision::DuplicateVoter;
138    }
139    if !target.version_compatible {
140        return VoterReconfigurationDecision::IncompatibleVoter;
141    }
142    if !target.caught_up {
143        return VoterReconfigurationDecision::VoterNotCaughtUp;
144    }
145    // A voter set of `usize::MAX` members cannot grow; fail closed.
146    let Some(next_voter_count) = voters.voter_count.checked_add(1) else {
147        return VoterReconfigurationDecision::InvalidVersionTransition;
148    };
149    admit_voter_set_change(next_voter_count, voters.kraft_version)
150}
151
152/// `RemoveVoterHandler`: version, then whether `VoterSet.removeVoter`
153/// returns a new set; an absent id, a different stored key, and the last
154/// voter are all its `VOTER_NOT_FOUND`.
155#[cfg_attr(creusot, requires(decision_voter_set_valid(voters)))]
156#[cfg_attr(
157    creusot,
158    ensures(voter_set_change_admission_valid(voters, VoterChangeKind::Remove, result,))
159)]
160#[cfg_attr(creusot, ensures(match result {
161    VoterReconfigurationDecision::Admit(_) =>
162        voter_set_removes(target.membership, voters),
163    VoterReconfigurationDecision::UnsupportedKraftVersion => true,
164    VoterReconfigurationDecision::VoterNotFound => voters.kraft_version@ == 1
165        && !voter_set_removes(target.membership, voters),
166    _ => false,
167}))]
168fn remove_decision(voters: CurrentVoterSet, target: TargetVoter) -> VoterReconfigurationDecision {
169    if voters.kraft_version != 1 {
170        return VoterReconfigurationDecision::UnsupportedKraftVersion;
171    }
172    if !matches!(target.membership, TargetMembership::PresentSameDirectory)
173        || voters.voter_count == 1
174    {
175        return VoterReconfigurationDecision::VoterNotFound;
176    }
177    admit_voter_set_change(voters.voter_count - 1, voters.kraft_version)
178}
179
180/// `UpdateVoterHandler`: `ApiVersions` range, then the voter lookup.
181#[cfg_attr(creusot, requires(decision_voter_set_valid(voters)))]
182#[cfg_attr(creusot, ensures(match result {
183    VoterReconfigurationDecision::Admit(plan) =>
184        target.version_compatible
185            && update_key_matches(target.membership)
186            && admitted_plan(voters, VoterChangeKind::Update, plan),
187    VoterReconfigurationDecision::IncompatibleVoter => !target.version_compatible,
188    VoterReconfigurationDecision::VoterNotFound => target.version_compatible
189        && target.membership == TargetMembership::Absent,
190    VoterReconfigurationDecision::DirectoryMismatch => target.version_compatible
191        && target.membership != TargetMembership::Absent
192        && !update_key_matches(target.membership),
193    _ => false,
194}))]
195fn update_decision(voters: CurrentVoterSet, target: TargetVoter) -> VoterReconfigurationDecision {
196    if !target.version_compatible {
197        return VoterReconfigurationDecision::IncompatibleVoter;
198    }
199    match target.membership {
200        TargetMembership::Absent => return VoterReconfigurationDecision::VoterNotFound,
201        TargetMembership::PresentOtherDirectory => {
202            return VoterReconfigurationDecision::DirectoryMismatch;
203        }
204        TargetMembership::PresentUnknownDirectory | TargetMembership::PresentSameDirectory => {}
205    }
206    let preflight_only = voters.kraft_version == 0;
207    VoterReconfigurationDecision::Admit(VoterReconfigurationPlan {
208        next_voter_count: voters.voter_count,
209        next_kraft_version: voters.kraft_version,
210        write_voters: !preflight_only,
211        write_kraft_version: false,
212        preflight_only,
213    })
214}
215
216/// `maybeAppendUpgradedKRaftVersion`: only the 0 to 1 upgrade, and only when
217/// every voter supports version 1.
218#[cfg_attr(creusot, requires(voters.voter_count@ > 0))]
219#[cfg_attr(creusot, ensures(match result {
220    VoterReconfigurationDecision::Admit(plan) =>
221        voters.kraft_version@ == 0
222            && request.requested_kraft_version@ == 1
223            && voters.all_voters_support_requested
224            && admitted_plan(voters, VoterChangeKind::FinalizeKraftVersion, plan),
225    VoterReconfigurationDecision::InvalidVersionTransition => voters.kraft_version@ != 0
226        || request.requested_kraft_version@ != 1,
227    VoterReconfigurationDecision::IncompatibleVoter => voters.kraft_version@ == 0
228        && request.requested_kraft_version@ == 1
229        && !voters.all_voters_support_requested,
230    _ => false,
231}))]
232fn finalize_decision(
233    voters: CurrentVoterSet,
234    request: VoterChangeRequest,
235) -> VoterReconfigurationDecision {
236    if voters.kraft_version != 0 || request.requested_kraft_version != 1 {
237        return VoterReconfigurationDecision::InvalidVersionTransition;
238    }
239    if !voters.all_voters_support_requested {
240        return VoterReconfigurationDecision::IncompatibleVoter;
241    }
242    VoterReconfigurationDecision::Admit(VoterReconfigurationPlan {
243        next_voter_count: voters.voter_count,
244        next_kraft_version: 1,
245        write_voters: true,
246        write_kraft_version: true,
247        preflight_only: false,
248    })
249}
250
251// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
252#[cfg(creusot)]
253#[cfg_attr(test, mutants::skip)]
254#[logic(open)]
255/// Decision helpers receive a nonempty set using a supported KRaft version.
256fn decision_voter_set_valid(voters: CurrentVoterSet) -> bool {
257    pearlite! { voters.voter_count@ > 0 && voters.kraft_version@ <= 1 }
258}
krabka_verified/reconfiguration/voter_reconfiguration_decision.rs

krabka_verified/reconfiguration/
voter_reconfiguration_decision.rs
