krabka_raft/kraft/transport/command.rs
1//! The command and inbound vocabulary that drives one turn of the engine loop.
2//!
3//! [`Command`] is everything that arrives on the engine's mpsc: an inbound peer
4//! RPC ([`Inbound`]), an injected core [`Event`], a decoded Fetch or
5//! `FetchSnapshot` response, a [`TimerTick`], or a handle-facing operation. The
6//! reply payloads that those handle ops carry back, [`MetadataFetchSlice`] and
7//! [`QuorumStateSnapshot`], live here with them.
8
9use {
10 crate::{
11 DelegationTokenMutation, SubmitChangeResult,
12 error::RaftError,
13 kraft::{
14 event::Event,
15 snapshot_fetch::SnapshotId,
16 transport::PeerApi,
17 types::{Epoch, NodeId, ReplicaKey},
18 },
19 reconfig::{ReconfigOutcome, VoterChange},
20 },
21 bytes::Bytes,
22 std::{collections::BTreeMap, sync::Arc},
23 tokio::{
24 sync::{mpsc, oneshot},
25 time::Instant,
26 },
27};
28
29/// An inbound KIP-595 RPC body, the request version its header carried, and a
30/// oneshot to reply on.
31///
32/// The event loop decodes the body at `version` into a core [`Event`], runs
33/// it, and encodes the produced response at the same version back onto
34/// `reply`, as Kafka answers every request at the version it arrived at.
35#[derive(Debug)]
36pub enum Inbound {
37 Vote {
38 req: Bytes,
39 version: i16,
40 reply: oneshot::Sender<Bytes>,
41 },
42 BeginQuorumEpoch {
43 req: Bytes,
44 version: i16,
45 reply: oneshot::Sender<Bytes>,
46 },
47 EndQuorumEpoch {
48 req: Bytes,
49 version: i16,
50 reply: oneshot::Sender<Bytes>,
51 },
52 Fetch {
53 req: Bytes,
54 version: i16,
55 reply: oneshot::Sender<Bytes>,
56 },
57 FetchSnapshot {
58 req: Bytes,
59 version: i16,
60 reply: oneshot::Sender<Bytes>,
61 },
62}
63
64impl Inbound {
65 /// The inbound RPC of peer api `api`.
66 #[must_use]
67 pub fn new(api: PeerApi, req: Bytes, version: i16, reply: oneshot::Sender<Bytes>) -> Self {
68 match api {
69 PeerApi::Fetch => Self::Fetch {
70 req,
71 version,
72 reply,
73 },
74 PeerApi::Vote => Self::Vote {
75 req,
76 version,
77 reply,
78 },
79 PeerApi::BeginQuorumEpoch => Self::BeginQuorumEpoch {
80 req,
81 version,
82 reply,
83 },
84 PeerApi::EndQuorumEpoch => Self::EndQuorumEpoch {
85 req,
86 version,
87 reply,
88 },
89 PeerApi::FetchSnapshot => Self::FetchSnapshot {
90 req,
91 version,
92 reply,
93 },
94 }
95 }
96
97 /// Whether the peer stopped waiting for the reply.
98 #[must_use]
99 pub fn caller_gone(&self) -> bool {
100 let (Self::Fetch { reply, .. }
101 | Self::Vote { reply, .. }
102 | Self::BeginQuorumEpoch { reply, .. }
103 | Self::EndQuorumEpoch { reply, .. }
104 | Self::FetchSnapshot { reply, .. }) = self;
105 reply.is_closed()
106 }
107
108 /// The peer api, request body, request version and reply oneshot.
109 #[must_use]
110 pub fn into_parts(self) -> (PeerApi, Bytes, i16, oneshot::Sender<Bytes>) {
111 let api = match &self {
112 Self::Fetch { .. } => PeerApi::Fetch,
113 Self::Vote { .. } => PeerApi::Vote,
114 Self::BeginQuorumEpoch { .. } => PeerApi::BeginQuorumEpoch,
115 Self::EndQuorumEpoch { .. } => PeerApi::EndQuorumEpoch,
116 Self::FetchSnapshot { .. } => PeerApi::FetchSnapshot,
117 };
118 let (Self::Fetch {
119 req,
120 version,
121 reply,
122 }
123 | Self::Vote {
124 req,
125 version,
126 reply,
127 }
128 | Self::BeginQuorumEpoch {
129 req,
130 version,
131 reply,
132 }
133 | Self::EndQuorumEpoch {
134 req,
135 version,
136 reply,
137 }
138 | Self::FetchSnapshot {
139 req,
140 version,
141 reply,
142 }) = self;
143 (api, req, version, reply)
144 }
145}
146
147/// A [`Command`] on the engine's mpsc, with the instant it was sent: the start
148/// of its Kafka `EventQueueTimeMs`.
149pub struct QueuedCommand {
150 pub command: Command,
151 pub enqueued_at: Instant,
152}
153
154/// The sending half of the engine's mpsc. It stamps each [`Command`] with
155/// the instant it was sent, before any wait for room in the queue, as Kafka
156/// stamps an event when it creates it.
157#[derive(Debug, Clone)]
158pub struct CommandSender(mpsc::Sender<QueuedCommand>);
159
160impl CommandSender {
161 /// A bounded command channel of `capacity`.
162 #[must_use]
163 pub fn channel(capacity: usize) -> (Self, mpsc::Receiver<QueuedCommand>) {
164 let (sender, receiver) = mpsc::channel(capacity);
165 (Self(sender), receiver)
166 }
167
168 /// Send `command` to the engine, waiting for room in the queue.
169 ///
170 /// # Errors
171 /// Returns the command back when the engine task is gone.
172 pub async fn send(&self, command: Command) -> Result<(), mpsc::error::SendError<Command>> {
173 let queued = QueuedCommand {
174 command,
175 enqueued_at: Instant::now(),
176 };
177 self.0
178 .send(queued)
179 .await
180 .map_err(|mpsc::error::SendError(queued)| mpsc::error::SendError(queued.command))
181 }
182
183 /// The room left in the queue.
184 #[must_use]
185 pub fn capacity(&self) -> usize { self.0.capacity() }
186}
187
188/// Everything that arrives on the engine's mpsc and drives one turn of the
189/// loop.
190pub enum Command {
191 /// An inbound peer RPC with a oneshot to reply on.
192 Inbound(Inbound),
193 /// Injects a core [`Event`] directly. This is the test and driver entry
194 /// point.
195 Event(Event),
196 /// A Fetch RESPONSE the follower received from the leader. A Fetch
197 /// response carries log records. The follower must truncate, append, and
198 /// apply those records BEFORE it feeds the `ReceiveFetchResponse` event to
199 /// the core, so this response gets its own command.
200 FetchResponse {
201 /// The leader that answered, which is the responder peer.
202 from: NodeId,
203 /// The raw encoded [`wire::PeerResponse::Fetch`](super::wire::PeerResponse::Fetch)
204 /// body.
205 body: Bytes,
206 },
207 /// A Vote RESPONSE a nominee received from a voter. Beside the grant it
208 /// names the voter's epoch, the leader it knows and that leader's
209 /// endpoint, which the engine acts on before the core counts the grant,
210 /// as Kafka's `maybeHandleCommonResponse` does.
211 VoteResponse { from: NodeId, body: Bytes },
212 /// A `FetchSnapshot` RESPONSE the follower received from the leader. It
213 /// carries snapshot bytes that the follower reassembles before it resumes.
214 /// This mirrors the dedicated command path of `FetchResponse`.
215 FetchSnapshotResponse { from: NodeId, body: Bytes },
216 /// An election, fetch, or heartbeat timer fired. The loop maps it to the
217 /// right core event after it reads the liveness state. A fetch tick
218 /// re-polls instead of electing, unless the leader has been missed enough
219 /// times.
220 Timer(TimerTick),
221 /// Handle op: append and commit a metadata batch as the leader. It replies
222 /// once the batch is committed and applied, or it replies with a
223 /// rejection.
224 SubmitChange {
225 records: Vec<krabka_metadata::MetadataRecord>,
226 expected_image: Option<Arc<krabka_metadata::MetadataImage>>,
227 reply: oneshot::Sender<Result<SubmitChangeResult, RaftError>>,
228 },
229 /// Handle op: compare and mutate delegation-token state in one engine turn.
230 SubmitDelegationTokenMutations {
231 mutations: Vec<DelegationTokenMutation>,
232 reply: oneshot::Sender<Result<SubmitChangeResult, RaftError>>,
233 },
234 /// Handle op: append a KIP-853 control batch and optionally wait for it to
235 /// commit under the new voter set.
236 Reconfigure {
237 change: VoterChange,
238 reply: oneshot::Sender<Result<ReconfigOutcome, RaftError>>,
239 },
240 /// Handle op: snapshot the current image to a checkpoint.
241 TriggerSnapshot {
242 reply: oneshot::Sender<Result<(), RaftError>>,
243 },
244 /// Handle op: read a structured snapshot of consensus state for
245 /// `DescribeQuorum`.
246 QuorumStateSnapshot {
247 reply: oneshot::Sender<QuorumStateSnapshot>,
248 },
249 /// Handle op: read a committed `__cluster_metadata` slice for an observer's
250 /// `API_KEY_METADATA_FETCH` (1004), encoded as Kafka record batches.
251 MetadataFetch {
252 fetch_offset: i64,
253 max_size: krabka_units::ByteSize,
254 /// The observer that asked, when it is a replica. A leader tracks its
255 /// progress under this key.
256 replica: Option<ReplicaKey>,
257 reply: oneshot::Sender<MetadataFetchSlice>,
258 },
259 /// Test-only: append a metadata batch to the log, the same way the
260 /// leader's `submit_change` does, and drive the commit through the real
261 /// apply pipeline. Replies with the appended base offset.
262 #[cfg(test)]
263 TestAppendAndCommit {
264 records: Vec<krabka_metadata::MetadataRecord>,
265 reply: oneshot::Sender<i64>,
266 },
267 /// Stop the loop gracefully: a leader resigns and the loop runs on, within
268 /// a bound, until the quorum has elected a successor.
269 Shutdown,
270 /// Stop the loop at once, handing nothing over, as a halted process does.
271 Halt,
272}
273
274impl Command {
275 /// Whether the caller of a command that replies stopped waiting for the
276 /// reply. A command that sends no reply never has a caller to lose.
277 #[must_use]
278 pub fn caller_gone(&self) -> bool {
279 match self {
280 Self::Inbound(inbound) => inbound.caller_gone(),
281 Self::SubmitChange { reply, .. }
282 | Self::SubmitDelegationTokenMutations { reply, .. } => reply.is_closed(),
283 Self::Reconfigure { reply, .. } => reply.is_closed(),
284 Self::TriggerSnapshot { reply } => reply.is_closed(),
285 Self::QuorumStateSnapshot { reply } => reply.is_closed(),
286 Self::MetadataFetch { reply, .. } => reply.is_closed(),
287 #[cfg(test)]
288 Self::TestAppendAndCommit { reply, .. } => reply.is_closed(),
289 Self::Event(_)
290 | Self::FetchResponse { .. }
291 | Self::VoteResponse { .. }
292 | Self::FetchSnapshotResponse { .. }
293 | Self::Timer(_)
294 | Self::Shutdown
295 | Self::Halt => false,
296 }
297 }
298}
299
300/// A committed-range read result for the observer metadata-fetch path (1004).
301///
302/// `records` is concatenated Kafka `RecordBatch`es, one for each committed log
303/// batch in `[fetch_offset, high_watermark)`. The offsets are `KraftLog`
304/// offsets. A `fetch_offset` below `log_start_offset` reads no records at all:
305/// the slice then carries `snapshot_id`, and the observer installs that
306/// snapshot before it fetches again.
307#[derive(Debug, Clone)]
308pub struct MetadataFetchSlice {
309 pub records: bytes::Bytes,
310 /// The KIP-630 snapshot `(end_offset, epoch)` that replaces `records` when
311 /// the observer asked for an offset this node has already pruned away.
312 /// `records` is empty whenever this is set.
313 pub snapshot_id: Option<SnapshotId>,
314 pub log_start_offset: i64,
315 pub high_watermark: i64,
316 /// Highest offset the quorum has committed, as this node last heard it.
317 ///
318 /// This node's own `high_watermark` bounds what it can serve; this says
319 /// how much more there is. They differ whenever the node answering the
320 /// fetch is itself a follower that is still catching up, and an observer
321 /// measures its own lag against this one.
322 pub quorum_high_watermark: i64,
323}
324
325/// Which timer fired. The loop interprets the tick against the current role
326/// and liveness state, and does not map it one-to-one to a core event.
327#[derive(Debug, Clone, Copy, PartialEq, Eq)]
328pub enum TimerTick {
329 /// The election timer's deadline passed.
330 Election,
331 /// The fetch timer's deadline passed. This is the follower and observer
332 /// poll watchdog.
333 Fetch,
334 /// The leader heartbeat interval ticked.
335 Heartbeat,
336 /// The leader's check-quorum deadline passed. This is the leader-side
337 /// watchdog: a majority of the voters has stopped fetching, so the leader
338 /// resigns instead of holding an epoch it has lost.
339 CheckQuorum,
340}
341
342/// One observer row of `DescribeQuorum`: a replica the leader has seen fetch
343/// that is not a voter, with what the leader tracks for it.
344///
345/// A timestamp is -1 while the leader has not recorded that event, as in Kafka.
346#[derive(Debug, Clone, PartialEq, Eq)]
347pub struct ObserverReplica {
348 pub id: NodeId,
349 /// The observer's directory id, nil when its fetches carry none.
350 pub directory_id: uuid::Uuid,
351 /// The offset of the observer's last valid fetch.
352 pub log_end_offset: i64,
353 /// Wall-clock milliseconds of its last valid fetch.
354 pub last_fetch_ms: i64,
355 /// Wall-clock milliseconds it was last caught up with the leader's log.
356 pub last_caught_up_ms: i64,
357}
358
359/// A structured, node-local snapshot of consensus state for the handle, which
360/// serves the broker's `DescribeQuorum` admin view.
361///
362/// This is the engine's own view. The handle maps it into the public
363/// `crate::controller::QuorumState`.
364#[derive(Debug, Clone)]
365pub struct QuorumStateSnapshot {
366 pub leader_id: Option<NodeId>,
367 pub leader_epoch: Epoch,
368 pub high_watermark: i64,
369 /// Append timestamp, in Unix epoch milliseconds, of the record at
370 /// `high_watermark - 1` when it is still available in the local log (or
371 /// the installed snapshot's last contained record timestamp). `0` means
372 /// no applied record has a known timestamp.
373 pub last_applied_record_timestamp_ms: i64,
374 /// Lifetime average elapsed microseconds of completed controller apply
375 /// passes that processed newly committed records.
376 pub pending_record_processing_time_us_avg: u64,
377 /// Lifetime maximum elapsed microseconds of completed controller apply
378 /// passes that processed newly committed records.
379 pub pending_record_processing_time_us_max: u64,
380 /// Highest offset the quorum has committed, as this node last heard it:
381 /// this node's own high watermark on a leader, and the leader-reported
382 /// high watermark on a follower that is still replaying the log.
383 ///
384 /// `high_watermark` alone cannot answer "how far behind is this node",
385 /// because a follower clamps it to its own log end. A follower that has
386 /// replicated 10 of the quorum's 10 000 committed records reports
387 /// `high_watermark == 10` and `quorum_high_watermark == 10_000`.
388 pub quorum_high_watermark: i64,
389 pub log_end_offset: i64,
390 /// Log-start offset. It rises past 0 once the log has been pruned below a
391 /// snapshot under KIP-630.
392 pub log_start_offset: i64,
393 pub voters: krabka_metadata::VoterSet,
394 /// Directory identity voted for in the current epoch, if any.
395 pub voted_directory_id: Option<uuid::Uuid>,
396 /// The replicas that fetch from this leader without holding a vote and
397 /// fetched within the last five minutes (Kafka's `observerStates`), one per
398 /// `(id, directory id)` key. Empty unless this node leads.
399 pub observers: Vec<ObserverReplica>,
400 /// Per-voter fetch offset, populated on the leader.
401 pub per_replica_fetch_offset: BTreeMap<NodeId, i64>,
402 /// Per-voter last fetch timestamp in wall-clock milliseconds.
403 pub per_replica_last_fetch_ms: BTreeMap<NodeId, i64>,
404 /// Per-voter last caught-up timestamp in wall-clock milliseconds.
405 pub per_replica_last_caught_up_ms: BTreeMap<NodeId, i64>,
406 /// Whether this node currently leads the metadata quorum.
407 pub is_leader: bool,
408 /// Raft consensus state name (leader, follower, candidate, observer).
409 pub current_state: &'static str,
410}