krabka_verified/isr/
watermark.rs

1use {
2    super::{
3        HighWatermarkFacts, HwmReplica, IsrCandidateFacts, IsrEligibilityFacts, IsrMemberRole,
4    },
5    creusot_std::prelude::*,
6};
7
8/// Kafka's `Partition.isReplicaIsrEligible`: the broker is neither fenced nor
9/// in controlled shutdown, and `isBrokerEpochIsrEligible` holds -- both epochs
10/// are known, and the Fetch carried either no epoch (-1) or the alive one.
11// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
12#[cfg(creusot)]
13#[cfg_attr(test, mutants::skip)]
14#[logic]
15fn kafka_is_replica_isr_eligible(facts: IsrEligibilityFacts) -> bool {
16    pearlite! {
17        !facts.fenced && !facts.shutting_down
18            && match facts.fetch_broker_epoch {
19                Some(stored) => match facts.alive_broker_epoch {
20                    Some(alive) => stored@ == -1 || stored@ == alive@,
21                    None => false,
22                },
23                None => false,
24            }
25    }
26}
27
28/// Kafka's `ReplicaState.isCaughtUp`: the follower's log end offset is the
29/// leader's, or it last caught up within `replica.lag.time.max.ms`. A
30/// follower that has not fetched from this leader has log end -1, Kafka's
31/// `UNKNOWN_OFFSET`, which no leader log end equals.
32// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
33#[cfg(creusot)]
34#[cfg_attr(test, mutants::skip)]
35#[logic]
36fn kafka_replica_is_caught_up(
37    follower_log_end: Int,
38    leader_log_end: Int,
39    caught_up_within_lag: bool,
40) -> bool {
41    pearlite! { follower_log_end == leader_log_end || caught_up_within_lag }
42}
43
44/// Decide whether one replica may join the ISR, Kafka's
45/// `Partition.isReplicaIsrEligible`.
46#[ensures(result == kafka_is_replica_isr_eligible(facts))]
47#[must_use]
48pub fn replica_isr_eligible(facts: IsrEligibilityFacts) -> bool {
49    if facts.fenced || facts.shutting_down {
50        return false;
51    }
52    match (facts.fetch_broker_epoch, facts.alive_broker_epoch) {
53        (Some(stored), Some(alive)) => stored == -1 || stored == alive,
54        _ => false,
55    }
56}
57
58/// Kafka's `Partition.isFollowerInSync`: the follower's log end offset has
59/// reached both the leader's high watermark and the start of the leader's
60/// current epoch. The second half keeps out a follower that could otherwise
61/// be elected before it fetched committed data between the high watermark
62/// and the leader's log end.
63// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
64#[cfg(creusot)]
65#[cfg_attr(test, mutants::skip)]
66#[logic]
67fn kafka_is_follower_in_sync(facts: IsrCandidateFacts) -> bool {
68    pearlite! {
69        facts.follower_log_end@ >= facts.leader_high_watermark@
70            && match facts.leader_epoch_start {
71                Some(start) => facts.follower_log_end@ >= start@,
72                None => false,
73            }
74    }
75}
76
77/// Decide whether one candidate belongs in the next ISR proposal.
78///
79/// The leader always stays, and a member that is no longer assigned leaves.
80/// An in-sync follower stays exactly while Kafka's `ReplicaState.isCaughtUp`
81/// holds, which is `Partition.getOutOfSyncReplicas`: a follower that keeps
82/// fetching but never catches up leaves once its last caught-up time is older
83/// than the bound. An out-of-sync follower joins exactly when Kafka's
84/// `Partition.needsExpandIsr` would add it: it is ISR-eligible
85/// ([`replica_isr_eligible`]) and in sync by `Partition.isFollowerInSync`.
86#[ensures(match facts.role {
87    IsrMemberRole::Unassigned => !result,
88    IsrMemberRole::Leader => result,
89    IsrMemberRole::InSyncFollower => result == kafka_replica_is_caught_up(
90        facts.follower_log_end@, facts.leader_log_end@, facts.caught_up_within_lag),
91    IsrMemberRole::OutOfSyncFollower => result == (
92        kafka_is_replica_isr_eligible(facts.eligibility) && kafka_is_follower_in_sync(facts)),
93})]
94#[must_use]
95pub fn isr_candidate_selected(facts: IsrCandidateFacts) -> bool {
96    match facts.role {
97        IsrMemberRole::Unassigned => false,
98        IsrMemberRole::Leader => true,
99        IsrMemberRole::InSyncFollower => {
100            facts.follower_log_end == facts.leader_log_end || facts.caught_up_within_lag
101        }
102        IsrMemberRole::OutOfSyncFollower => {
103            let in_sync = facts.follower_log_end >= facts.leader_high_watermark
104                && matches!(
105                    facts.leader_epoch_start,
106                    Some(start) if facts.follower_log_end >= start
107                );
108            in_sync && replica_isr_eligible(facts.eligibility)
109        }
110    }
111}
112
113/// Kafka's `Partition.isUnderMinIsr` on the leader.
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 kafka_is_under_min_isr(facts: HighWatermarkFacts) -> bool {
119    pearlite! { facts.isr_size@ < facts.effective_min_isr@ }
120}
121
122/// Whether `replica` holds the high watermark back: it is in the ISR, or, in
123/// Kafka's `shouldWaitForReplicaToJoinIsr`, it is caught up and ISR-eligible,
124/// so the watermark waits for it rather than run away from a follower that is
125/// about to join.
126// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
127#[cfg(creusot)]
128#[cfg_attr(test, mutants::skip)]
129#[logic]
130fn kafka_holds_watermark(replica: HwmReplica, leader_log_end: Int) -> bool {
131    pearlite! {
132        replica.in_isr
133            || (kafka_replica_is_caught_up(replica.log_end@, leader_log_end, replica.caught_up_within_lag)
134                && kafka_is_replica_isr_eligible(replica.eligibility))
135    }
136}
137
138/// The candidate Kafka's `maybeIncrementLeaderHW` loop builds over the first
139/// `n` replicas: the leader's log end, lowered to every log end of a replica
140/// that holds the watermark back.
141// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
142#[cfg(creusot)]
143#[cfg_attr(test, mutants::skip)]
144#[logic]
145#[variant(n)]
146fn kafka_watermark_candidate(leader_log_end: Int, replicas: Seq<HwmReplica>, n: Int) -> Int {
147    pearlite! {
148        if n <= 0 {
149            leader_log_end
150        } else {
151            if kafka_holds_watermark(replicas[n - 1], leader_log_end)
152                && replicas[n - 1].log_end@ < kafka_watermark_candidate(leader_log_end, replicas, n - 1)
153            {
154                replicas[n - 1].log_end@
155            } else {
156                kafka_watermark_candidate(leader_log_end, replicas, n - 1)
157            }
158        }
159    }
160}
161
162/// Kafka's `Partition.maybeIncrementLeaderHW` followed by
163/// `UnifiedLog.maybeIncrementHighWatermark`: nothing moves while the ISR is
164/// under min ISR, and otherwise the watermark rises to the candidate if that
165/// is higher and never falls.
166// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
167#[cfg(creusot)]
168#[cfg_attr(test, mutants::skip)]
169#[logic]
170fn kafka_leader_high_watermark(facts: HighWatermarkFacts, replicas: Seq<HwmReplica>) -> Int {
171    pearlite! {
172        if kafka_is_under_min_isr(facts) {
173            facts.current@
174        } else if kafka_watermark_candidate(facts.leader_log_end@, replicas, replicas.len())
175            > facts.current@
176        {
177            kafka_watermark_candidate(facts.leader_log_end@, replicas, replicas.len())
178        } else {
179            facts.current@
180        }
181    }
182}
183
184/// Compute the leader's next high watermark.
185///
186/// While the ISR is below the effective `min.insync.replicas` the watermark
187/// stays where it is, so every record below it reached the HWM with at least
188/// min ISR replicas holding it -- the fact KIP-966 builds the eligible-leader
189/// set on. Otherwise it is the lowest log end among the leader, the ISR and
190/// every caught-up ISR-eligible replica outside it, if that is higher than
191/// the current watermark. A replica that has not fetched from this leader
192/// (log end -1) in the ISR holds it where it is.
193///
194/// The caller keeps `current` at or below the leader's log end, as Kafka's
195/// `UnifiedLog.truncateTo` does when it truncates.
196#[requires(facts.current@ <= facts.leader_log_end@)]
197#[ensures(result@ == kafka_leader_high_watermark(facts, replicas@))]
198#[ensures(facts.current@ <= result@ && result@ <= facts.leader_log_end@)]
199#[ensures(kafka_is_under_min_isr(facts) ==> result == facts.current)]
200#[must_use]
201pub fn leader_high_watermark(facts: HighWatermarkFacts, replicas: &[HwmReplica]) -> i64 {
202    if facts.isr_size < facts.effective_min_isr {
203        return facts.current;
204    }
205    let mut candidate = facts.leader_log_end;
206    let mut i = 0;
207    #[cfg_attr(creusot, invariant(i@ <= [email protected]()))]
208    #[cfg_attr(creusot, invariant(candidate@ <= facts.leader_log_end@))]
209    #[cfg_attr(creusot, invariant(candidate@
210        == kafka_watermark_candidate(facts.leader_log_end@, replicas@, i@)))]
211    #[cfg_attr(creusot, variant([email protected]() - i@))]
212    while i < replicas.len() {
213        let Some(&replica) = replicas.get(i) else {
214            break;
215        };
216        let caught_up = replica.log_end == facts.leader_log_end || replica.caught_up_within_lag;
217        let holds = replica.in_isr || (caught_up && replica_isr_eligible(replica.eligibility));
218        if holds && replica.log_end < candidate {
219            candidate = replica.log_end;
220        }
221        i += 1;
222    }
223    if candidate > facts.current {
224        candidate
225    } else {
226        facts.current
227    }
228}
229
230/// Report a proposal only when at least one unique member is added or removed.
231#[ensures(result == (removed@ > 0 || added@ > 0))]
232#[must_use]
233pub fn isr_proposal_changed(removed: usize, added: usize) -> bool { removed > 0 || added > 0 }
krabka_verified/isr/watermark.rs

krabka_verified/isr/
watermark.rs
