krabka_verified/consensus/
failover.rs

1use {
2    super::{FailoverAction, FailoverFacts, FailoverRecovery, LiveIsr, RecoveryCandidate},
3    creusot_std::prelude::*,
4};
5
6/// Majority size for a voter set: `floor(n / 2) + 1`.
7#[ensures(result@ == voter_count@ / 2 + 1)]
8#[must_use]
9pub const fn majority_size(voter_count: usize) -> usize { voter_count / 2 + 1 }
10
11/// Whether the unique grants from current voters reach a majority.
12#[ensures(result == (current_grants@ >= voter_count@ / 2 + 1))]
13#[must_use]
14pub const fn election_has_quorum(voter_count: usize, current_grants: usize) -> bool {
15    current_grants >= majority_size(voter_count)
16}
17
18/// Select the failover action class by a fixed precedence ladder.
19///
20/// What the contract proves is the ladder, and only the ladder. For a dead
21/// leader it picks, in order: a clean election when a live ISR member can
22/// lead; otherwise, and only when no ISR member is live, an election from the
23/// eligible leader replicas; otherwise the last known leader; otherwise the
24/// configured offset-aware recovery; otherwise the KIP-841 election when it is
25/// available; otherwise the partition stays unavailable. A live ISR of
26/// witnesses only stops the ladder at unavailable. For a live leader it only
27/// shrinks the ISR or leaves it alone. Each outcome is pinned to exactly one
28/// combination of [`FailoverFacts`].
29///
30/// Why each rung is safe is not part of the proof. It rests on the host's
31/// classification and on Kafka's semantics: an ISR member holds every
32/// committed record; a KIP-966 eligible leader replica left the ISR while the
33/// partition still held `min.insync.replicas` members, so it holds every
34/// committed record too, and neither the `unclean.leader.election.enable`
35/// toggle nor `unclean.recovery.strategy` gates it. Its rung is reachable only
36/// once the live ISR is empty, the same guard Apache Kafka's
37/// `PartitionChangeBuilder.isValidNewLeader` puts on its `targetElr` disjunct.
38/// The last known leader is the one replica that led when the partition lost
39/// its leader, but it may have lost an unflushed tail, so electing it is an
40/// unclean election that Kafka takes without either toggle: its rung sits
41/// directly under the ELR one, exactly where `electAnyLeader` and
42/// `electPreferredLeader` put `canElectLastKnownLeader`, ahead of the
43/// KIP-841 branch.
44///
45/// `unclean_election_available` joins the KIP-841 toggle and the existence of
46/// a replica that can serve, because the ladder never separates them: an
47/// election that is allowed with nobody to elect and one that has a candidate
48/// and no permission are the same unavailable partition.
49#[ensures(match result {
50    FailoverAction::ElectClean => facts.leader_dead && facts.live_isr == LiveIsr::Electable,
51    FailoverAction::ElectFromElr => facts.leader_dead
52        && facts.live_isr == LiveIsr::Empty
53        && facts.out_of_isr.has_electable_elr,
54    FailoverAction::ElectLastKnown => facts.leader_dead
55        && facts.live_isr == LiveIsr::Empty
56        && !facts.out_of_isr.has_electable_elr
57        && facts.out_of_isr.last_known_leader_electable,
58    FailoverAction::Recover(selected) => facts.leader_dead
59        && facts.live_isr == LiveIsr::Empty
60        && !facts.out_of_isr.has_electable_elr
61        && !facts.out_of_isr.last_known_leader_electable
62        && facts.out_of_isr.recovery != FailoverRecovery::None
63        && selected == facts.out_of_isr.recovery,
64    FailoverAction::ElectUnclean => facts.leader_dead
65        && facts.live_isr == LiveIsr::Empty
66        && !facts.out_of_isr.has_electable_elr
67        && !facts.out_of_isr.last_known_leader_electable
68        && facts.out_of_isr.recovery == FailoverRecovery::None
69        && facts.out_of_isr.unclean_election_available,
70    FailoverAction::Unavailable => facts.leader_dead
71        && (facts.live_isr == LiveIsr::WitnessesOnly
72            || (facts.live_isr == LiveIsr::Empty
73                && !facts.out_of_isr.has_electable_elr
74                && !facts.out_of_isr.last_known_leader_electable
75                && facts.out_of_isr.recovery == FailoverRecovery::None
76                && !facts.out_of_isr.unclean_election_available)),
77    FailoverAction::ShrinkIsr => !facts.leader_dead && facts.isr_shrunk,
78    FailoverAction::NoChange => !facts.leader_dead && !facts.isr_shrunk,
79})]
80#[must_use]
81pub fn failover_action(facts: FailoverFacts) -> FailoverAction {
82    if !facts.leader_dead {
83        return if facts.isr_shrunk {
84            FailoverAction::ShrinkIsr
85        } else {
86            FailoverAction::NoChange
87        };
88    }
89    match facts.live_isr {
90        LiveIsr::Electable => return FailoverAction::ElectClean,
91        LiveIsr::WitnessesOnly => return FailoverAction::Unavailable,
92        LiveIsr::Empty => {}
93    }
94    let out_of_isr = facts.out_of_isr;
95    if out_of_isr.has_electable_elr {
96        return FailoverAction::ElectFromElr;
97    }
98    if out_of_isr.last_known_leader_electable {
99        return FailoverAction::ElectLastKnown;
100    }
101    match out_of_isr.recovery {
102        FailoverRecovery::Balanced | FailoverRecovery::Aggressive => {
103            FailoverAction::Recover(out_of_isr.recovery)
104        }
105        FailoverRecovery::None if out_of_isr.unclean_election_available => {
106            FailoverAction::ElectUnclean
107        }
108        FailoverRecovery::None => FailoverAction::Unavailable,
109    }
110}
111
112/// `a` ranks at or above `b` as a KIP-966 recovery candidate: the higher last
113/// written leader epoch first, then the higher log end offset, then the lower
114/// broker ID.
115// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
116#[cfg(creusot)]
117#[cfg_attr(test, mutants::skip)]
118#[logic]
119pub fn ranks_ge(a: RecoveryCandidate, b: RecoveryCandidate) -> bool {
120    pearlite! {
121        a.last_epoch@ > b.last_epoch@
122            || (a.last_epoch@ == b.last_epoch@
123                && (a.log_end_offset@ > b.log_end_offset@
124                    || (a.log_end_offset@ == b.log_end_offset@ && a.broker_id@ <= b.broker_id@)))
125    }
126}
127
128/// Select the replica with highest `(last leader epoch, log end offset)` and
129/// lowest broker ID as the deterministic tie-breaker.
130///
131/// The contract states that ranking once, as `ranks_ge`: the returned index
132/// ranks at or above every candidate. `None` is returned exactly for an empty
133/// slice.
134#[ensures(match result {
135    None => [email protected]() == 0,
136    Some(best) => best@ < [email protected]()
137        && forall<j: Int> 0 <= j && j < [email protected]()
138            ==> ranks_ge(candidates@[best@], candidates@[j]),
139})]
140#[must_use]
141pub fn select_best_recovery_replica(candidates: &[RecoveryCandidate]) -> Option<usize> {
142    let n = candidates.len();
143    if n == 0 {
144        return None;
145    }
146    let mut best = 0usize;
147    let mut i = 1usize;
148    #[invariant(1 <= i@ && i@ <= n@)]
149    #[invariant(best@ < i@)]
150    #[invariant(forall<j: Int> 0 <= j && j < i@ ==> ranks_ge(candidates@[best@], candidates@[j]))]
151    #[variant(n@ - i@)]
152    while i < n {
153        if let (Some(candidate), Some(current)) = (candidates.get(i), candidates.get(best))
154            && (candidate.last_epoch > current.last_epoch
155                || (candidate.last_epoch == current.last_epoch
156                    && candidate.log_end_offset > current.log_end_offset)
157                || (candidate.last_epoch == current.last_epoch
158                    && candidate.log_end_offset == current.log_end_offset
159                    && candidate.broker_id < current.broker_id))
160        {
161            best = i;
162        }
163        i += 1;
164    }
165    Some(best)
166}
167
168// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
169#[cfg(creusot)]
170#[cfg_attr(test, mutants::skip)]
171#[logic(open)]
172/// Members of `{log_end} U s` with value >= `v`. This is the
173/// majority-replication witness.
174pub fn hwm_member_at(log_end: Int, s: Seq<i64>, k: Int) -> Int {
175    pearlite! {
176        if k == 0 { log_end } else { s[k - 1]@ }
177    }
178}
179
180// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
181#[cfg(creusot)]
182#[cfg_attr(test, mutants::skip)]
183#[logic(open)]
184#[variant(limit)]
185pub fn count_ge_prefix(log_end: Int, s: Seq<i64>, v: Int, limit: Int, leader_counts: bool) -> Int {
186    pearlite! {
187        if limit <= 0 {
188            0
189        } else {
190            count_ge_prefix(log_end, s, v, limit - 1, leader_counts)
191                + (if (leader_counts || limit > 1)
192                    && hwm_member_at(log_end, s, limit - 1) >= v { 1 } else { 0 })
193        }
194    }
195}
196
197// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
198#[cfg(creusot)]
199#[cfg_attr(test, mutants::skip)]
200#[logic(open)]
201#[variant(s.len())]
202pub fn count_ge(log_end: Int, s: Seq<i64>, v: Int, leader_counts: bool) -> Int {
203    pearlite! { count_ge_prefix(log_end, s, v, s.len() + 1, leader_counts) }
204}
205
206// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
207#[cfg(creusot)]
208#[cfg_attr(test, mutants::skip)]
209#[logic]
210#[requires(0 <= limit && limit <= left.len() + 1 && limit <= right.len() + 1)]
211#[requires(forall<i: Int> 0 <= i && i < limit - 1
212    ==> (left[i]@ >= threshold) == (right[i]@ >= threshold))]
213#[ensures(count_ge_prefix(log_end, left, threshold, limit, false)
214    == count_ge_prefix(log_end, right, threshold, limit, false))]
215#[variant(limit)]
216pub fn lemma_explicit_vote_count_equal(
217    log_end: Int,
218    left: Seq<i64>,
219    right: Seq<i64>,
220    threshold: Int,
221    limit: Int,
222) {
223    if limit > 0 {
224        lemma_explicit_vote_count_equal(log_end, left, right, threshold, limit - 1);
225    }
226}
krabka_verified/consensus/failover.rs

krabka_verified/consensus/
failover.rs
