krabka_verified/isr/
admission.rs

1use {
2    super::{
3        AlterPartitionFacts, CaughtUpCredit, IsrAdmission, IsrMemberFacts, IsrMemberRole,
4        ProposedIsr,
5    },
6    creusot_std::prelude::*,
7};
8
9/// Kafka's `ReplicationControlManager.validateAlterPartitionData`, check for
10/// check, after its unknown-partition check.
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_validate_alter_partition_data(facts: AlterPartitionFacts) -> IsrAdmission {
16    pearlite! {
17        if facts.request_leader_epoch@ > facts.current_leader_epoch@
18            || facts.request_partition_epoch@ > facts.current_partition_epoch@
19        {
20            IsrAdmission::NotController
21        } else if facts.request_leader_epoch@ < facts.current_leader_epoch@ {
22            IsrAdmission::FencedLeaderEpoch
23        } else if !facts.requester_is_leader {
24            IsrAdmission::InvalidRequest
25        } else if facts.request_partition_epoch@ < facts.current_partition_epoch@ {
26            IsrAdmission::InvalidUpdateVersion
27        } else {
28            match facts.proposed_isr {
29                ProposedIsr::Valid => {
30                    if !facts.recovery_state_valid {
31                        IsrAdmission::InvalidRequest
32                    } else if !facts.replicas_eligible {
33                        IsrAdmission::IneligibleReplica
34                    } else {
35                        IsrAdmission::Admit
36                    }
37                }
38                _ => IsrAdmission::InvalidRequest,
39            }
40        }
41    }
42}
43
44/// Apply Kafka's ordered `AlterPartition` checks to one partition row.
45///
46/// A row is admitted exactly when it names the controller's current leader and
47/// partition epochs, comes from the leader, proposes a valid ISR that keeps
48/// the leader, requests a legal recovery state, and names only eligible
49/// replicas. In particular a row whose partition epoch is behind the
50/// controller's never overwrites the newer ISR.
51#[ensures(result == kafka_validate_alter_partition_data(facts))]
52#[ensures((result == IsrAdmission::Admit) == (
53    facts.request_leader_epoch@ == facts.current_leader_epoch@
54        && facts.request_partition_epoch@ == facts.current_partition_epoch@
55        && facts.requester_is_leader
56        && facts.proposed_isr == ProposedIsr::Valid
57        && facts.recovery_state_valid
58        && facts.replicas_eligible
59))]
60#[must_use]
61pub fn isr_admission(facts: AlterPartitionFacts) -> IsrAdmission {
62    if facts.request_leader_epoch > facts.current_leader_epoch
63        || facts.request_partition_epoch > facts.current_partition_epoch
64    {
65        return IsrAdmission::NotController;
66    }
67    if facts.request_leader_epoch < facts.current_leader_epoch {
68        return IsrAdmission::FencedLeaderEpoch;
69    }
70    if !facts.requester_is_leader {
71        return IsrAdmission::InvalidRequest;
72    }
73    if facts.request_partition_epoch < facts.current_partition_epoch {
74        return IsrAdmission::InvalidUpdateVersion;
75    }
76    match facts.proposed_isr {
77        ProposedIsr::Valid => {}
78        ProposedIsr::Invalid | ProposedIsr::WithoutLeader => {
79            return IsrAdmission::InvalidRequest;
80        }
81    }
82    if !facts.recovery_state_valid {
83        return IsrAdmission::InvalidRequest;
84    }
85    if facts.replicas_eligible {
86        IsrAdmission::Admit
87    } else {
88        IsrAdmission::IneligibleReplica
89    }
90}
91
92/// Kafka's `Replica.updateFetchStateOrThrow` choice of `lastCaughtUpTimeMs`.
93///
94/// `fetch_offset` is the follower's fetch offset, `leader_log_end` the
95/// leader's log end offset when the fetch arrived, and
96/// `last_fetch_leader_log_end` the leader's log end offset recorded at the
97/// follower's previous fetch (-1 before the first). The host keeps the later
98/// of the credited time and the recorded one.
99#[ensures((result == CaughtUpCredit::ThisFetch) == (fetch_offset@ >= leader_log_end@))]
100#[ensures((result == CaughtUpCredit::PreviousFetch) == (
101    fetch_offset@ < leader_log_end@ && fetch_offset@ >= last_fetch_leader_log_end@
102))]
103#[must_use]
104pub fn follower_caught_up_credit(
105    fetch_offset: i64,
106    leader_log_end: i64,
107    last_fetch_leader_log_end: i64,
108) -> CaughtUpCredit {
109    if fetch_offset >= leader_log_end {
110        CaughtUpCredit::ThisFetch
111    } else if fetch_offset >= last_fetch_leader_log_end {
112        CaughtUpCredit::PreviousFetch
113    } else {
114        CaughtUpCredit::Unchanged
115    }
116}
117
118/// Kafka's `ReplicaState.isCaughtUp`: the follower has the leader's log end
119/// offset, or it last caught up within `replica.lag.time.max.ms`.
120// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
121#[cfg(creusot)]
122#[cfg_attr(test, mutants::skip)]
123#[logic]
124fn kafka_is_caught_up(facts: IsrMemberFacts) -> bool {
125    pearlite! { facts.log_end_matches_leader || facts.caught_up_within_lag }
126}
127
128/// krabka's former ISR scan rule, superseded by [`crate::isr::isr_candidate_selected`].
129///
130/// The leader always stays. An in-sync follower stays exactly while Kafka's
131/// `ReplicaState.isCaughtUp` holds. An out-of-sync follower is admitted when
132/// it fetched and caught up within the bound, which is not Kafka's
133/// `Partition.isFollowerInSync`. The broker no longer calls this; the one
134/// remaining caller is the leader-failover model's expansion stand-in, and
135/// this function and [`IsrMemberFacts`] go once that model moves to
136/// [`crate::isr::isr_candidate_selected`].
137#[ensures(match facts.role {
138    IsrMemberRole::Unassigned => !result,
139    IsrMemberRole::Leader => result,
140    IsrMemberRole::InSyncFollower => result == kafka_is_caught_up(facts),
141    IsrMemberRole::OutOfSyncFollower =>
142        result == (facts.fetch_within_lag && facts.caught_up_within_lag),
143})]
144#[must_use]
145pub fn isr_maintenance_selected(facts: IsrMemberFacts) -> bool {
146    match facts.role {
147        IsrMemberRole::Unassigned => false,
148        IsrMemberRole::Leader => true,
149        IsrMemberRole::InSyncFollower => facts.log_end_matches_leader || facts.caught_up_within_lag,
150        IsrMemberRole::OutOfSyncFollower => facts.fetch_within_lag && facts.caught_up_within_lag,
151    }
152}
krabka_verified/isr/admission.rs

krabka_verified/isr/
admission.rs
