krabka_verified/
isr.rs

1//! Classic Kafka ISR decisions: `AlterPartition` validation on the controller,
2//! ISR maintenance and follower catch-up tracking on the partition leader, and
3//! the high-watermark computation.
4
5use creusot_std::prelude::*;
6
7/// How a proposed ISR relates to the partition's assignment and leader.
8#[krabka_macros::model_type]
9pub enum ProposedIsr {
10    /// Kafka's `Replicas.validateIsr` fails: a member is negative, repeated,
11    /// or not an assigned replica.
12    Invalid,
13    /// Every member is a distinct assigned replica, but the current leader is
14    /// not among them.
15    WithoutLeader,
16    /// Every member is a distinct assigned replica, and the leader is one.
17    Valid,
18}
19
20/// What the controller knows about one `AlterPartition` partition row, as
21/// Kafka's `ReplicationControlManager.validateAlterPartitionData` reads it.
22#[krabka_macros::model_type]
23pub struct AlterPartitionFacts {
24    pub request_leader_epoch: i32,
25    pub current_leader_epoch: i32,
26    pub request_partition_epoch: i32,
27    pub current_partition_epoch: i32,
28    /// The request's `brokerId` is the partition's current leader.
29    pub requester_is_leader: bool,
30    pub proposed_isr: ProposedIsr,
31    /// The requested leader recovery state is a known value, a `RECOVERING`
32    /// request proposes at most one member, and a `RECOVERED` partition is not
33    /// asked to go back to `RECOVERING`.
34    pub recovery_state_valid: bool,
35    /// Kafka's `ineligibleReplicasForIsr` is empty: every proposed member is
36    /// registered, unfenced, not in controlled shutdown, and carries either
37    /// the -1 sentinel or its registered broker epoch.
38    pub replicas_eligible: bool,
39}
40
41/// The first `AlterPartition` validation failure of one partition row, or
42/// `Admit`.
43#[krabka_macros::model_type]
44pub enum IsrAdmission {
45    /// `NOT_CONTROLLER`: the request carries a leader or partition epoch the
46    /// controller has not seen, so this controller is likely stale.
47    NotController,
48    /// `FENCED_LEADER_EPOCH`: the request's leader epoch is older.
49    FencedLeaderEpoch,
50    /// `INVALID_UPDATE_VERSION`: the request's partition epoch is older.
51    InvalidUpdateVersion,
52    /// `INVALID_REQUEST`: the requester is not the leader, the proposed ISR is
53    /// invalid or omits the leader, or the recovery-state change is illegal.
54    InvalidRequest,
55    /// `INELIGIBLE_REPLICA`: a proposed member may not join an ISR.
56    IneligibleReplica,
57    Admit,
58}
59
60/// Which fetch time, if any, a follower fetch proves the follower was caught
61/// up at.
62#[krabka_macros::model_type]
63pub enum CaughtUpCredit {
64    /// The fetch reached the leader's log end offset: caught up now.
65    ThisFetch,
66    /// The fetch reached the leader's log end offset as of the previous fetch:
67    /// caught up at the previous fetch's time.
68    PreviousFetch,
69    /// The fetch proves nothing new.
70    Unchanged,
71}
72
73/// Where one candidate stands in the partition the leader maintains.
74#[krabka_macros::model_type]
75pub enum IsrMemberRole {
76    /// A current ISR member that is no longer assigned.
77    Unassigned,
78    /// The assigned leader.
79    Leader,
80    /// An assigned follower in the current ISR.
81    InSyncFollower,
82    /// An assigned follower outside the current ISR.
83    OutOfSyncFollower,
84}
85
86/// The inputs of [`isr_maintenance_selected`], superseded by
87/// [`IsrCandidateFacts`]. Every time bound is `replica.lag.time.max.ms`,
88/// measured from the scan's single timestamp.
89#[krabka_macros::model_type]
90pub struct IsrMemberFacts {
91    pub role: IsrMemberRole,
92    /// The follower's log end offset equals the leader's at this scan.
93    pub log_end_matches_leader: bool,
94    /// The follower's last caught-up time is within the bound.
95    pub caught_up_within_lag: bool,
96    /// The follower's last fetch is within the bound.
97    pub fetch_within_lag: bool,
98}
99
100/// What the leader's metadata cache and the follower's last Fetch say about
101/// whether one replica may join the ISR, the inputs of Kafka's
102/// `Partition.isReplicaIsrEligible` (KIP-841).
103#[krabka_macros::model_type]
104pub struct IsrEligibilityFacts {
105    /// Kafka's `metadataCache.isBrokerFenced`.
106    pub fenced: bool,
107    /// Kafka's `metadataCache.isBrokerShuttingDown`.
108    pub shutting_down: bool,
109    /// Kafka's `ReplicaState.brokerEpoch`: the broker epoch the follower's
110    /// last Fetch to this leader carried, -1 when that Fetch carried none, and
111    /// `None` before this leader received one.
112    pub fetch_broker_epoch: Option<i64>,
113    /// Kafka's `metadataCache.getAliveBrokerEpoch`: the broker's registered
114    /// epoch while it is registered and unfenced, `None` otherwise.
115    pub alive_broker_epoch: Option<i64>,
116}
117
118/// What the leader knows about one candidate at an ISR maintenance scan,
119/// read once at the scan's single timestamp.
120#[krabka_macros::model_type]
121pub struct IsrCandidateFacts {
122    pub role: IsrMemberRole,
123    /// Kafka's `ReplicaState.logEndOffset`: the follower's last fetch offset,
124    /// -1 before it fetched from this leader.
125    pub follower_log_end: i64,
126    /// The leader's log end offset.
127    pub leader_log_end: i64,
128    /// The leader's high watermark.
129    pub leader_high_watermark: i64,
130    /// Kafka's `Partition.leaderEpochStartOffsetOpt`: the log end offset at
131    /// which this leader's epoch began, `None` while unknown.
132    pub leader_epoch_start: Option<i64>,
133    /// The follower's last caught-up time is within
134    /// `replica.lag.time.max.ms`.
135    pub caught_up_within_lag: bool,
136    pub eligibility: IsrEligibilityFacts,
137}
138
139/// One remote replica as Kafka's `Partition.maybeIncrementLeaderHW` weighs
140/// it.
141#[krabka_macros::model_type]
142pub struct HwmReplica {
143    /// Kafka's `ReplicaState.logEndOffset`, -1 before the replica fetched from
144    /// this leader.
145    pub log_end: i64,
146    /// The replica is in the leader's ISR.
147    pub in_isr: bool,
148    /// The replica's last caught-up time is within `replica.lag.time.max.ms`.
149    pub caught_up_within_lag: bool,
150    pub eligibility: IsrEligibilityFacts,
151}
152
153/// The partition-wide inputs of Kafka's `Partition.maybeIncrementLeaderHW`.
154#[krabka_macros::model_type]
155pub struct HighWatermarkFacts {
156    /// The leader's high watermark before this step.
157    pub current: i64,
158    /// The leader's log end offset.
159    pub leader_log_end: i64,
160    /// Kafka's `partitionState.isr.size`: the ISR, leader included.
161    pub isr_size: usize,
162    /// Kafka's `Partition.effectiveMinIsr`: `min.insync.replicas` capped by
163    /// the replica count.
164    pub effective_min_isr: usize,
165}
166
167mod admission;
168pub use admission::{follower_caught_up_credit, isr_admission, isr_maintenance_selected};
169
170mod watermark;
171pub use watermark::{
172    isr_candidate_selected, isr_proposal_changed, leader_high_watermark, replica_isr_eligible,
173};
174
175#[cfg(test)]
176mod tests;
krabka_verified/isr.rs

krabka_verified/
isr.rs
