krabka_verified/
registration.rs

1//! Broker heartbeat admission decisions.
2
3use creusot_std::prelude::*;
4
5/// How one broker heartbeat relates to the broker's registration.
6///
7/// Fencing and shutdown are not decided here: the controller's heartbeat
8/// state machine derives them from the current broker state, as Kafka's
9/// `BrokerHeartbeatManager.calculateNextBrokerState` does.
10#[krabka_macros::model_type]
11pub enum BrokerHeartbeatDecision {
12    /// No registration exists for the broker id.
13    Missing,
14    /// The heartbeat carries another broker epoch.
15    Stale,
16    /// The heartbeat carries the registered epoch. The broker has caught up
17    /// once its metadata offset reaches its registration record, whose offset
18    /// is the broker epoch.
19    Current { caught_up: bool },
20}
21
22/// Fence an absent or stale registration, as Kafka's
23/// `ClusterControlManager.checkBrokerEpoch` does, and otherwise report whether
24/// the broker has replayed its own registration record.
25#[ensures(match result {
26    BrokerHeartbeatDecision::Missing => registered_epoch == None,
27    BrokerHeartbeatDecision::Stale => exists<epoch: i64>
28        registered_epoch == Some(epoch) && request_epoch != epoch,
29    BrokerHeartbeatDecision::Current { caught_up } => exists<epoch: i64>
30        registered_epoch == Some(epoch)
31            && request_epoch == epoch
32            && caught_up == (metadata_offset@ >= epoch@),
33})]
34#[must_use]
35pub fn broker_heartbeat_decision(
36    registered_epoch: Option<i64>,
37    request_epoch: i64,
38    metadata_offset: i64,
39) -> BrokerHeartbeatDecision {
40    match registered_epoch {
41        None => BrokerHeartbeatDecision::Missing,
42        Some(epoch) if request_epoch != epoch => BrokerHeartbeatDecision::Stale,
43        Some(epoch) => BrokerHeartbeatDecision::Current {
44            caught_up: metadata_offset >= epoch,
45        },
46    }
47}
48
49#[cfg(test)]
50mod tests {
51    use {
52        super::{BrokerHeartbeatDecision as D, broker_heartbeat_decision},
53        assert2::assert,
54    };
55
56    #[test]
57    fn heartbeat_fences_exact_epochs_and_reports_catch_up() {
58        let cases = [
59            ("unregistered broker", None, 7, 7, D::Missing),
60            ("older epoch", Some(7), 6, i64::MAX, D::Stale),
61            ("newer epoch", Some(7), 8, i64::MAX, D::Stale),
62            ("behind its registration", Some(7), 7, 6, D::Current {
63                caught_up: false,
64            }),
65            ("at its registration", Some(7), 7, 7, D::Current {
66                caught_up: true,
67            }),
68            ("past its registration", Some(7), 7, 9, D::Current {
69                caught_up: true,
70            }),
71        ];
72        for (case, registered_epoch, request_epoch, metadata_offset, expected) in cases {
73            assert!(
74                broker_heartbeat_decision(registered_epoch, request_epoch, metadata_offset)
75                    == expected,
76                "{case}"
77            );
78        }
79    }
80}
krabka_verified/registration.rs

krabka_verified/
registration.rs
