krabka_client_coordination/transport.rs
1//! The seam between the succession rules and the cluster.
2//!
3//! [`CoordinationTransport`] holds every operation that the succession rules
4//! need from a Kafka cluster, and it holds nothing else. The rules decide who
5//! registers, who challenges, and when. This trait performs the five requests
6//! that carry those decisions to the brokers.
7//!
8//! The seam exists so a test drives the rules without a broker.
9//! `#[mockall::automock]` generates `MockCoordinationTransport` under
10//! `cfg(test)`, and a unit test scripts each method with the record sequence
11//! it wants. [`broker::BrokerTransport`] is the one implementation that
12//! talks to a cluster.
13//!
14//! # Which calls carry authority
15//!
16//! [`CoordinationTransport::acquire_epoch`] and
17//! [`CoordinationTransport::write_lease`] are the guarded pair.
18//! `acquire_epoch` mints the epoch of the role and fences the member that held
19//! it before. `write_lease` writes under that epoch, and the broker rejects
20//! the write when a later member has taken the role. A deposed holder learns
21//! that it lost the role from [`CoordinationError::Fenced`], and from nothing
22//! else.
23//!
24//! [`CoordinationTransport::register`] carries no authority. A candidate holds
25//! no epoch when it announces itself, so the registration is a plain append.
26//! [`CoordinationTransport::read_role_records`] and
27//! [`CoordinationTransport::describe`] read.
28
29use async_trait::async_trait;
30
31use crate::{
32 error::CoordinationError,
33 record::{CoordinationKey, CoordinationRecord, FencingToken, Lease, MemberId, Role},
34};
35
36/// The records of one role's partition, in offset order.
37///
38/// Each entry pairs the offset of the record with its decoded key and value.
39/// The offset is the join sequence of a registration, because compaction keeps
40/// the offset of every record it retains. The succession rules rank candidates
41/// on that offset.
42pub type RoleRecords = Vec<(i64, CoordinationKey, CoordinationRecord)>;
43
44/// The cluster operations that the coordination client performs.
45///
46/// A caller that wants the real cluster uses
47/// [`BrokerTransport`](crate::broker::BrokerTransport). A test uses the generated
48/// `MockCoordinationTransport`.
49#[cfg_attr(test, mockall::automock)]
50#[async_trait]
51pub trait CoordinationTransport: Send + Sync {
52 /// Mints a new epoch for `role` and fences the member that held it.
53 ///
54 /// The transaction coordinator picks the epoch, so the value is
55 /// quorum-minted and monotonic. The implementation keeps the writer that
56 /// the epoch binds, and [`Self::write_lease`] uses it.
57 ///
58 /// # Errors
59 ///
60 /// Returns [`CoordinationError::Producer`] when the coordinator refuses
61 /// the call, and [`CoordinationError::Client`] when the connection to the
62 /// coordinator fails.
63 async fn acquire_epoch(&self, role: &Role) -> Result<FencingToken, CoordinationError>;
64
65 /// Reads the whole partition of `role` and returns the records in offset
66 /// order.
67 ///
68 /// The read takes committed records only, so an aborted lease write is
69 /// invisible. The result holds the records of `role` and drops every
70 /// record of another role that shares the partition.
71 ///
72 /// # Errors
73 ///
74 /// Returns [`CoordinationError::Client`] when a fetch fails, and
75 /// [`CoordinationError::Record`] when a record of `role` does not decode.
76 async fn read_role_records(&self, role: &Role) -> Result<RoleRecords, CoordinationError>;
77
78 /// Appends the registration of `member` to the partition of `role`.
79 ///
80 /// A candidate holds no epoch, so this append sits outside a transaction
81 /// and carries no authority. Its offset is the join sequence that the
82 /// succession rules rank on.
83 ///
84 /// # Errors
85 ///
86 /// Returns [`CoordinationError::Producer`] when the append fails, and
87 /// [`CoordinationError::Client`] when the connection fails.
88 async fn register(&self, role: &Role, member: &MemberId) -> Result<(), CoordinationError>;
89
90 /// Writes the lease of `role` in a transaction under `token`.
91 ///
92 /// The broker rejects the write when a later member has taken `role`, and
93 /// that rejection is how a deposed holder learns it lost the role.
94 ///
95 /// # Errors
96 ///
97 /// Returns [`CoordinationError::Fenced`] when the broker rejects `token`.
98 /// Returns [`CoordinationError::NotHeld`] when this transport never minted
99 /// `token`. Returns [`CoordinationError::Producer`] for every other
100 /// failure of the write.
101 async fn write_lease(
102 &self,
103 role: &Role,
104 token: FencingToken,
105 lease: &Lease,
106 ) -> Result<(), CoordinationError>;
107
108 /// Asks the transaction coordinator which token holds `role` now.
109 ///
110 /// The result is `None` when no member has ever held `role`. A third party
111 /// calls this to check the authority of a writer. It joins no group and it
112 /// takes no epoch.
113 ///
114 /// # Errors
115 ///
116 /// Returns [`CoordinationError::Admin`] when the coordinator lookup fails
117 /// or the coordinator reports a fault other than an unknown role.
118 async fn describe(&self, role: &Role) -> Result<Option<FencingToken>, CoordinationError>;
119}
120
121#[cfg(test)]
122mod tests {
123 use assert2::{assert, check};
124
125 use super::*;
126 use crate::record::{CoordinationRecord, Registration};
127
128 fn role() -> Role {
129 Role::new("controller").expect("a valid role")
130 }
131
132 fn member() -> MemberId {
133 MemberId::new("node-1").expect("a valid member id")
134 }
135
136 fn lease(token: FencingToken) -> Lease {
137 Lease {
138 member: member(),
139 token,
140 granted_at: 1_700_000_000_000,
141 deadline: 1_700_000_030_000,
142 }
143 }
144
145 fn token() -> FencingToken {
146 FencingToken::new(4242, 7).expect("a valid token")
147 }
148
149 /// The mock stands in for a cluster, so the succession rules and the
150 /// client layer both test without a broker. This test drives every method
151 /// of the seam through the mock.
152 #[tokio::test]
153 async fn the_mock_transport_answers_every_call_of_the_seam() {
154 let mut transport = MockCoordinationTransport::new();
155 transport
156 .expect_acquire_epoch()
157 .returning(|_role| Ok(token()));
158 transport
159 .expect_register()
160 .returning(|_role, _member| Ok(()));
161 transport
162 .expect_write_lease()
163 .returning(|_role, _token, _lease| Ok(()));
164 transport
165 .expect_describe()
166 .returning(|_role| Ok(Some(token())));
167 transport.expect_read_role_records().returning(|_role| {
168 Ok(vec![(
169 7,
170 CoordinationKey::registration(role(), member()),
171 CoordinationRecord::Registration(Registration {
172 member: member(),
173 registered_at: 1_700_000_000_000,
174 }),
175 )])
176 });
177
178 check!(transport.acquire_epoch(&role()).await.unwrap() == token());
179 check!(transport.register(&role(), &member()).await.is_ok());
180 check!(
181 transport
182 .write_lease(&role(), token(), &lease(token()))
183 .await
184 .is_ok()
185 );
186 check!(transport.describe(&role()).await.unwrap() == Some(token()));
187 let records = transport.read_role_records(&role()).await.unwrap();
188 check!(records.len() == 1);
189 check!(records[0].0 == 7);
190 }
191
192 /// A deposed holder learns that it lost the role from the error of
193 /// `write_lease`. The mock reproduces that shape, so the succession rules
194 /// test the loss path without a broker.
195 #[tokio::test]
196 async fn the_mock_transport_reports_a_fence_on_a_lease_write() {
197 let mut transport = MockCoordinationTransport::new();
198 transport
199 .expect_write_lease()
200 .returning(|role, _token, _lease| {
201 Err(CoordinationError::Fenced { role: role.clone() })
202 });
203
204 let outcome = transport
205 .write_lease(&role(), token(), &lease(token()))
206 .await;
207 let error = outcome.expect_err("the write reports a fence");
208 assert!(let CoordinationError::Fenced { .. } = &error);
209 check!(error.is_fenced());
210 check!(error.to_string() == "fenced: another member holds role controller");
211 }
212}