krabka_verified/broker.rs
1//! Pure, safety-critical decision kernels used by `krabka-broker`.
2//!
3//! Keeping these small arithmetic decisions here lets Creusot prove the exact
4//! executable bodies used by the asynchronous broker.
5
6use creusot_std::prelude::*;
7
8// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
9#[cfg(creusot)]
10#[cfg_attr(test, mutants::skip)]
11#[logic(open)]
12/// Truncation preserves the floor and clamps each visibility frontier to its end.
13pub(crate) fn clamped_fetch_watermarks(
14 original: FetchWatermarks,
15 bounded: FetchWatermarks,
16) -> bool {
17 pearlite! {
18 bounded.log_start == original.log_start
19 && bounded.hw@ == [email protected](bounded.log_end@)
20 && bounded.lso@ == [email protected](bounded.log_end@)
21 && bounded.deliverable@ == [email protected](bounded.log_end@)
22 }
23}
24
25// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
26#[cfg(creusot)]
27#[cfg_attr(test, mutants::skip)]
28#[logic(open)]
29/// Export the committed response watermarks and empty-view flags from one floor.
30pub(crate) fn committed_fetch_response(
31 view: FetchVisibility,
32 hw: i64,
33 lso: i64,
34 deliverable: i64,
35 floor: Int,
36 log_end: Int,
37) -> bool {
38 pearlite! {
39 view.response_hw == hw && view.response_lso@ == [email protected](lso@)
40 && view.effective_lso@ == [email protected](lso@) && view.read_committed_aborts
41 && view.out_of_range == (floor > log_end)
42 && view.empty == (floor <= log_end && floor >= [email protected](deliverable@))
43 }
44}
45
46/// Visibility bounds and response watermarks for one Fetch partition.
47#[krabka_macros::model_type]
48pub struct FetchVisibility {
49 pub out_of_range: bool,
50 pub empty: bool,
51 pub limit_offset: i64,
52 pub effective_lso: i64,
53 pub read_committed_aborts: bool,
54 pub response_hw: i64,
55 pub response_lso: i64,
56}
57
58/// The partition offsets one Fetch visibility decision reads.
59///
60/// They are one struct because they are five `i64` values with five different
61/// meanings, and a transposed call site would compile.
62#[krabka_macros::model_type]
63pub struct FetchWatermarks {
64 /// First offset the log still holds. Below it a fetch is out of range.
65 pub log_start: i64,
66 /// High watermark: the exclusive bound of what the ISR has replicated.
67 pub hw: i64,
68 /// Last stable offset: the first offset an open transaction may cover.
69 pub lso: i64,
70 /// Log end offset: the exclusive bound of what the leader holds.
71 pub log_end: i64,
72 /// KFC-1 delivery watermark: the first offset that is not due yet.
73 pub deliverable: i64,
74}
75
76/// The only direct mutation class selected from one follower Fetch response.
77#[krabka_macros::model_type]
78pub enum ReplicaFetchMutation {
79 /// The response is not for the exact live request target.
80 Reject,
81 /// The response is fenced or otherwise unsuccessful; error handling may
82 /// retry or enter a separately guarded recovery path.
83 Retry,
84 /// Apply the KIP-320 divergence boundary and return without appending.
85 Truncate,
86 /// Apply the successful response batches, then its high watermark.
87 Append,
88}
89
90/// What a follower knows about one row of a Fetch response when it decides
91/// what that row may change.
92///
93/// The row's topic and partition identity is not among them: the host looks
94/// the row up by that identity (`ResponseIndex::locate` in the broker's
95/// replicator) and never hands the kernel a row it could not attribute to
96/// exactly one followed partition.
97#[krabka_macros::model_type]
98pub struct ReplicaFetchFacts {
99 /// The leader epoch this follower sent for the partition in the request
100 /// that the row answers.
101 pub request_leader_epoch: i32,
102 /// The leader epoch of the partition's current replication target.
103 pub current_leader_epoch: i32,
104 /// The replication target is the one the request was sent to.
105 pub target_matches: bool,
106 /// The row's `current_leader` is either absent or exactly the current
107 /// target's leader and epoch.
108 pub reported_target_matches: bool,
109 /// The row's Kafka error code; `0` is `NONE`.
110 pub error_code: i16,
111 /// The row's KIP-320 `DivergingEpoch.Epoch`; `-1`, the schema default,
112 /// means the row carries no divergence. Kafka's
113 /// `FetchResponse.isDivergingEpoch` tests the epoch, not the end offset.
114 pub diverging_epoch: i32,
115}
116
117/// What a preferred-leader rebalance scan observed about one change it wants
118/// to submit.
119///
120/// The host reads every fact off the change record itself and the scan's
121/// liveness and witness snapshots, not off the election that produced it, so
122/// the kernel checks the election's output rather than restating it.
123#[krabka_macros::model_type]
124pub struct PreferredLeaderChange {
125 /// The leader the change installs.
126 pub new_leader: u64,
127 /// The partition's preferred replica, `replicas[0]`, or `None` for an
128 /// empty assignment.
129 pub preferred_replica: Option<u64>,
130 /// The new leader is in the change's ISR.
131 pub leader_in_isr: bool,
132 /// The new leader is alive in the scan's liveness snapshot.
133 pub leader_alive: bool,
134 /// The new leader carries the witness role, so it may never lead.
135 pub leader_is_witness: bool,
136}
137
138/// Offset facts used to admit one `DeleteRecords` trim.
139#[krabka_macros::model_type]
140pub struct DeleteRecordsTrimFacts {
141 pub requested: i64,
142 pub high_watermark: i64,
143 pub log_end: i64,
144 pub current_start: i64,
145 pub has_delivery_watermark: bool,
146 pub delivery_watermark: i64,
147}
148
149/// The complete boundary decision for one `DeleteRecords` trim.
150#[krabka_macros::model_type]
151pub enum DeleteRecordsTrimDecision {
152 RejectMalformed,
153 RejectOutOfRange,
154 Noop { frontier: i64 },
155 Apply { frontier: i64 },
156}
157
158/// The next idempotent step while reconciling WAL and local trim frontiers.
159#[krabka_macros::model_type]
160pub enum DeleteRecordsTrimApplication {
161 RejectMalformed,
162 TrimWal { frontier: i64 },
163 TrimLocal { frontier: i64 },
164 Complete { frontier: i64 },
165}
166
167/// The complete per-key admission outcome for `FindCoordinator`.
168///
169/// The allow variants preserve the key type for the host adapter. Denials carry
170/// the Kafka authorization domain, while malformed SHARE keys and unknown wire
171/// discriminants fail closed as invalid requests.
172#[krabka_macros::model_type]
173pub enum FindCoordinatorAdmission {
174 AllowGroup,
175 AllowTransaction,
176 AllowShare,
177 DenyGroup,
178 DenyTransaction,
179 DenyCluster,
180 InvalidRequest,
181}
182
183mod fetch;
184#[cfg(creusot)]
185pub use fetch::{fetch_limit_model, offset_min, preferred_change_eligible, replica_fetch_fenced};
186pub use fetch::{fetch_visibility, preferred_rebalance_admission, replica_fetch_mutation};
187
188mod trim_admission;
189#[cfg(creusot)]
190pub use trim_admission::effective_share_backlog_model;
191pub use trim_admission::{
192 delete_records_trim_application, delete_records_trim_decision, effective_share_backlog,
193};
194
195mod coordinator_partition;
196#[cfg(creusot)]
197pub use coordinator_partition::java_string_hash_prefix_model;
198pub use coordinator_partition::{
199 find_coordinator_admission, java_string_hash_partition, unclean_recovery_commit_admission,
200};
201
202#[cfg(test)]
203mod tests;