krabka_raft/kraft/transport/wire/
request.rs

1//! The `PeerRequest` body codec.
2//!
3//! `PeerRequest` is the flat request vocabulary the engine reasons in. This
4//! module maps each variant onto the generated KIP-595 request message at its
5//! captured version, and decodes each one back.
6
7use {
8    super::codec::{
9        FETCH_SNAPSHOT_VERSION, FETCH_VERSION, METADATA_PARTITION, METADATA_TOPIC,
10        METADATA_TOPIC_ID, QUORUM_EPOCH_VERSION, VOTE_VERSION, cluster_id_to_wire, encode_body,
11        epoch_from_wire, epoch_to_wire, node_from_wire, node_to_wire, uuid_from_wire, uuid_to_wire,
12    },
13    crate::{
14        kraft::{
15            snapshot_fetch::SnapshotId,
16            types::{Epoch, NodeId},
17        },
18        voter_wire::endpoint as voter_endpoint,
19    },
20    base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD},
21    bytes::Bytes,
22    krabka_metadata::VoterEndpoint,
23    krabka_protocol::{
24        Decode,
25        owned::{
26            begin_quorum_epoch_request::{self as bqe_req, BeginQuorumEpochRequest},
27            end_quorum_epoch_request::{self as eqe_req, EndQuorumEpochRequest},
28            fetch_request::{self as fetch_req, FetchRequest},
29            fetch_snapshot_request::{self as fs_req, FetchSnapshotRequest},
30            vote_request::{self as vote_req, VoteRequest},
31        },
32    },
33    krabka_verified::{
34        VoteWireDecision,
35        vote::{VoteEncodeDecision, vote_encode_decision},
36        vote_wire_decision,
37    },
38};
39
40fn begin_metadata_topic(partition: bqe_req::PartitionData) -> bqe_req::TopicData {
41    bqe_req::TopicData {
42        topic_name: METADATA_TOPIC.to_owned(),
43        partitions: vec![partition],
44        ..Default::default()
45    }
46}
47
48fn end_metadata_topic(partition: eqe_req::PartitionData) -> eqe_req::TopicData {
49    eqe_req::TopicData {
50        topic_name: METADATA_TOPIC.to_owned(),
51        partitions: vec![partition],
52        ..Default::default()
53    }
54}
55
56#[cfg(test)]
57mod tests;
58
59/// A peer RPC request body, as encoded by the sending engine and decoded by
60/// the receiving engine's inbound dispatch.
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub enum PeerRequest {
63    Vote {
64        cluster_id: Option<uuid::Uuid>,
65        /// The recipient voter this request is addressed to. This is the
66        /// wire top-level `voterId`. The JVM validates that an incoming
67        /// Vote is addressed to it before it considers the grant, and it
68        /// silently rejects a stale `voterId` or a `voterId` of `-1`.
69        /// `broadcast_vote` builds this field for each recipient.
70        voter_id: NodeId,
71        voter_directory_id: uuid::Uuid,
72        candidate_epoch: Epoch,
73        candidate: NodeId,
74        candidate_directory_id: uuid::Uuid,
75        last_epoch: Epoch,
76        last_offset: i64,
77        pre_vote: bool,
78    },
79    BeginQuorumEpoch {
80        /// The sender's cluster id, which the recipient checks.
81        cluster_id: Option<uuid::Uuid>,
82        /// The recipient voter key: `VoterId` and `VoterDirectoryId`, which
83        /// the recipient checks against its own (`isValidVoterKey`).
84        /// `broadcast_begin_quorum_epoch` builds it for each recipient.
85        voter_id: NodeId,
86        voter_directory_id: uuid::Uuid,
87        leader_id: NodeId,
88        leader_epoch: Epoch,
89        /// The leader's own listeners: `LeaderEndpoints`, which a recipient
90        /// whose voter set does not name the leader dials it at.
91        leader_endpoints: Vec<VoterEndpoint>,
92    },
93    EndQuorumEpoch {
94        /// The sender's cluster id, which the recipient checks.
95        cluster_id: Option<uuid::Uuid>,
96        leader_id: NodeId,
97        leader_epoch: Epoch,
98        /// The voters in order of replication progress, most caught up first,
99        /// each with its directory id: `PreferredCandidates` (KIP-853).
100        preferred_candidates: Vec<PreferredCandidate>,
101    },
102    Fetch {
103        /// The sender's cluster id, which the leader checks.
104        cluster_id: Option<uuid::Uuid>,
105        from: NodeId,
106        /// How long the leader may hold the request when it has nothing new:
107        /// `MaxWaitMs`.
108        max_wait_ms: i32,
109        /// The epoch the sender is in: `CurrentLeaderEpoch`, which the
110        /// responder checks in `validateLeaderOnlyRequest`. Raw, as for
111        /// `FetchSnapshot`, so a negative epoch stays fenced.
112        current_leader_epoch: i32,
113        /// The epoch of the sender's last fetched record: `LastFetchedEpoch`,
114        /// which the leader checks for divergence.
115        fetch_epoch: Epoch,
116        fetch_offset: i64,
117        replica_directory_id: uuid::Uuid,
118        /// The sender's high watermark, or -1 while it knows none:
119        /// `HighWatermark` (v18). The leader answers at once when its own is
120        /// higher.
121        high_watermark: i64,
122    },
123    FetchSnapshot {
124        /// The sender's cluster id, which the leader checks.
125        cluster_id: Option<uuid::Uuid>,
126        from: NodeId,
127        /// The epoch the sender believes the leader holds:
128        /// `CurrentLeaderEpoch`, which the leader checks against its own.
129        current_leader_epoch: i32,
130        snapshot_id: SnapshotId,
131        position: i64,
132        max_bytes: i32,
133    },
134}
135
136/// A voter that an `EndQuorumEpoch` names as a preferred successor.
137#[derive(Debug, Clone, PartialEq, Eq)]
138pub struct PreferredCandidate {
139    /// The voter's node id.
140    pub id: NodeId,
141    /// The voter's directory id.
142    pub directory_id: uuid::Uuid,
143}
144
145impl PeerRequest {
146    /// Encode a request whose Vote fields have already passed host validation.
147    ///
148    /// Test-only: production sends use [`Self::try_encode`] and fail closed on
149    /// a Vote identity or epoch past Kafka's signed `int32` range.
150    ///
151    /// # Panics
152    ///
153    /// Panics on a Vote whose identity or epoch does not fit `int32`.
154    #[cfg(test)]
155    #[must_use]
156    pub fn encode(&self) -> Bytes {
157        self.try_encode()
158            .expect("Vote fields must fit Kafka signed int32 wire fields")
159    }
160
161    /// Encode after checking the Vote identity and epoch wire range.
162    #[must_use]
163    pub fn try_encode(&self) -> Option<Bytes> {
164        match *self {
165            Self::Vote {
166                cluster_id,
167                voter_id,
168                voter_directory_id,
169                candidate_epoch,
170                candidate,
171                candidate_directory_id,
172                last_epoch,
173                last_offset,
174                pre_vote,
175            } => {
176                if vote_encode_decision(voter_id.0, candidate.0, candidate_epoch, last_epoch)
177                    != VoteEncodeDecision::Accept
178                {
179                    return None;
180                }
181                let req = VoteRequest {
182                    cluster_id: cluster_id_to_wire(cluster_id),
183                    voter_id: i32::try_from(voter_id.0).ok()?,
184                    topics: vec![vote_req::TopicData {
185                        topic_name: METADATA_TOPIC.to_owned(),
186                        partitions: vec![vote_req::PartitionData {
187                            partition_index: METADATA_PARTITION,
188                            replica_epoch: i32::try_from(candidate_epoch).ok()?,
189                            replica_id: i32::try_from(candidate.0).ok()?,
190                            replica_directory_id: uuid_to_wire(candidate_directory_id),
191                            voter_directory_id: uuid_to_wire(voter_directory_id),
192                            last_offset_epoch: i32::try_from(last_epoch).ok()?,
193                            last_offset,
194                            pre_vote,
195                            ..Default::default()
196                        }],
197                        ..Default::default()
198                    }],
199                    ..Default::default()
200                };
201                Some(encode_body(&req, VOTE_VERSION))
202            }
203            Self::BeginQuorumEpoch {
204                cluster_id,
205                voter_id,
206                voter_directory_id,
207                leader_id,
208                leader_epoch,
209                ref leader_endpoints,
210            } => {
211                // Kafka's `RaftUtil.singletonBeginQuorumEpochRequest`.
212                let req = BeginQuorumEpochRequest {
213                    cluster_id: cluster_id_to_wire(cluster_id),
214                    voter_id: node_to_wire(voter_id),
215                    leader_endpoints: leader_endpoints
216                        .iter()
217                        .map(|endpoint| bqe_req::LeaderEndpoint {
218                            name: endpoint.name.clone(),
219                            host: endpoint.host.clone(),
220                            port: endpoint.port,
221                            ..Default::default()
222                        })
223                        .collect(),
224                    topics: vec![begin_metadata_topic(bqe_req::PartitionData {
225                        partition_index: METADATA_PARTITION,
226                        voter_directory_id: uuid_to_wire(voter_directory_id),
227                        leader_id: node_to_wire(leader_id),
228                        leader_epoch: epoch_to_wire(leader_epoch),
229                        ..Default::default()
230                    })],
231                    ..Default::default()
232                };
233                Some(encode_body(&req, QUORUM_EPOCH_VERSION))
234            }
235            Self::EndQuorumEpoch {
236                cluster_id,
237                leader_id,
238                leader_epoch,
239                ref preferred_candidates,
240            } => {
241                // Kafka's `RaftUtil.singletonEndQuorumEpochRequest`, which
242                // names the successors both ways: `PreferredSuccessors` for
243                // v0, `PreferredCandidates` from v1.
244                let req = EndQuorumEpochRequest {
245                    cluster_id: cluster_id_to_wire(cluster_id),
246                    topics: vec![end_metadata_topic(eqe_req::PartitionData {
247                        partition_index: METADATA_PARTITION,
248                        leader_id: node_to_wire(leader_id),
249                        leader_epoch: epoch_to_wire(leader_epoch),
250                        preferred_successors: preferred_candidates
251                            .iter()
252                            .map(|candidate| node_to_wire(candidate.id))
253                            .collect(),
254                        preferred_candidates: preferred_candidates
255                            .iter()
256                            .map(|candidate| eqe_req::ReplicaInfo {
257                                candidate_id: node_to_wire(candidate.id),
258                                candidate_directory_id: uuid_to_wire(candidate.directory_id),
259                                ..Default::default()
260                            })
261                            .collect(),
262                        ..Default::default()
263                    })],
264                    ..Default::default()
265                };
266                Some(encode_body(&req, QUORUM_EPOCH_VERSION))
267            }
268            Self::Fetch {
269                cluster_id,
270                from,
271                max_wait_ms,
272                current_leader_epoch,
273                fetch_epoch,
274                fetch_offset,
275                replica_directory_id,
276                high_watermark,
277            } => {
278                // Kafka's `KafkaRaftClient.buildFetchRequest`.
279                let req = FetchRequest {
280                    cluster_id: cluster_id_to_wire(cluster_id),
281                    replica_id: node_to_wire(from),
282                    max_wait_ms,
283                    min_bytes: 1,
284                    max_bytes: 1024 * 1024,
285                    replica_state: fetch_req::ReplicaState {
286                        replica_id: node_to_wire(from),
287                        ..Default::default()
288                    },
289                    topics: vec![fetch_req::FetchTopic {
290                        topic: METADATA_TOPIC.to_owned(),
291                        topic_id: METADATA_TOPIC_ID,
292                        partitions: vec![fetch_req::FetchPartition {
293                            partition: METADATA_PARTITION,
294                            current_leader_epoch,
295                            fetch_offset,
296                            last_fetched_epoch: epoch_to_wire(fetch_epoch),
297                            replica_directory_id: uuid_to_wire(replica_directory_id),
298                            high_watermark,
299                            ..Default::default()
300                        }],
301                        ..Default::default()
302                    }],
303                    ..Default::default()
304                };
305                Some(encode_body(&req, FETCH_VERSION))
306            }
307            Self::FetchSnapshot {
308                cluster_id,
309                from,
310                current_leader_epoch,
311                snapshot_id,
312                position,
313                max_bytes,
314            } => Some(encode_fetch_snapshot_request(
315                cluster_id,
316                from,
317                current_leader_epoch,
318                snapshot_id,
319                ChunkWindow {
320                    position,
321                    max_bytes,
322                },
323            )),
324        }
325    }
326
327    /// Decodes a request body. Returns `None` on a malformed frame.
328    #[must_use]
329    pub fn decode(buf: &[u8]) -> Option<Self> { decode_vote(buf) }
330}
331
332/// Decodes a request body at `version`. The inbound handlers run Kafka's
333/// request checks on the result, so this does not refuse a field value: only a
334/// body that does not decode gives `None`. Kafka's
335/// `RequestContext.parseRequest` fails such a request, and the connection
336/// closes. Like that parser, this ignores bytes after the message.
337pub fn decode_request<T>(buf: &[u8], version: i16) -> Option<T>
338where
339    T: for<'de> Decode<'de>,
340{
341    let mut cur = buf;
342    T::decode(&mut cur, version).ok()
343}
344
345/// Kafka's `FetchRequest.replicaId`: the top-level `ReplicaId` up to v14, the
346/// `ReplicaState` from v15.
347#[must_use]
348pub fn fetch_replica_id(request: &FetchRequest, version: i16) -> i32 {
349    if version >= 15 {
350        request.replica_state.replica_id
351    } else {
352        request.replica_id
353    }
354}
355
356/// Decodes a Vote request body (api 52).
357#[must_use]
358pub fn decode_vote(buf: &[u8]) -> Option<PeerRequest> {
359    let mut cur = buf;
360    let req = VoteRequest::decode(&mut cur, VOTE_VERSION).ok()?;
361    if !cur.is_empty() || req.topics.len() != 1 {
362        return None;
363    }
364    let topic = req.topics.first()?;
365    if topic.topic_name != METADATA_TOPIC || topic.partitions.len() != 1 {
366        return None;
367    }
368    let p = topic.partitions.first()?;
369    if p.partition_index != METADATA_PARTITION
370        || vote_wire_decision(
371            req.voter_id,
372            p.replica_id,
373            p.replica_epoch,
374            p.last_offset_epoch,
375        ) != VoteWireDecision::Accept
376    {
377        return None;
378    }
379    Some(PeerRequest::Vote {
380        cluster_id: request_cluster_id(req.cluster_id.as_deref()).ok()?,
381        voter_id: NodeId(u64::try_from(req.voter_id).ok()?),
382        voter_directory_id: uuid_from_wire(p.voter_directory_id),
383        candidate_epoch: u32::try_from(p.replica_epoch).ok()?,
384        candidate: NodeId(u64::try_from(p.replica_id).ok()?),
385        candidate_directory_id: uuid_from_wire(p.replica_directory_id),
386        last_epoch: u32::try_from(p.last_offset_epoch).ok()?,
387        last_offset: p.last_offset,
388        pre_vote: p.pre_vote,
389    })
390}
391
392/// Parses an optional request `ClusterId`. An absent one is `Ok(None)`; one
393/// that is present but does not parse is the error.
394fn request_cluster_id(value: Option<&str>) -> Result<Option<uuid::Uuid>, ()> {
395    value.map(|id| parse_cluster_id(id).ok_or(())).transpose()
396}
397
398/// Parses a request `ClusterId`: Kafka's base64 form, or the hyphenated form.
399pub fn parse_cluster_id(value: &str) -> Option<uuid::Uuid> {
400    uuid::Uuid::parse_str(value).ok().or_else(|| {
401        let bytes: [u8; 16] = URL_SAFE_NO_PAD.decode(value).ok()?.try_into().ok()?;
402        Some(uuid::Uuid::from_bytes(bytes))
403    })
404}
405
406/// Decodes a `BeginQuorumEpoch` request body (api 53).
407#[must_use]
408pub fn decode_begin(buf: &[u8]) -> Option<PeerRequest> {
409    let req = decode_request::<BeginQuorumEpochRequest>(buf, QUORUM_EPOCH_VERSION)?;
410    let p = req.topics.first()?.partitions.first()?;
411    Some(PeerRequest::BeginQuorumEpoch {
412        cluster_id: request_cluster_id(req.cluster_id.as_deref()).ok()?,
413        voter_id: node_from_wire(req.voter_id),
414        voter_directory_id: uuid_from_wire(p.voter_directory_id),
415        leader_id: node_from_wire(p.leader_id),
416        leader_epoch: epoch_from_wire(p.leader_epoch),
417        leader_endpoints: req
418            .leader_endpoints
419            .iter()
420            .map(|endpoint| voter_endpoint(&endpoint.name, &endpoint.host, endpoint.port))
421            .collect(),
422    })
423}
424
425/// Decodes an `EndQuorumEpoch` request body (api 54).
426#[must_use]
427pub fn decode_end(buf: &[u8]) -> Option<PeerRequest> {
428    let req = decode_request::<EndQuorumEpochRequest>(buf, QUORUM_EPOCH_VERSION)?;
429    let p = req.topics.first()?.partitions.first()?;
430    Some(PeerRequest::EndQuorumEpoch {
431        cluster_id: request_cluster_id(req.cluster_id.as_deref()).ok()?,
432        leader_id: node_from_wire(p.leader_id),
433        leader_epoch: epoch_from_wire(p.leader_epoch),
434        preferred_candidates: p
435            .preferred_candidates
436            .iter()
437            .map(|candidate| PreferredCandidate {
438                id: node_from_wire(candidate.candidate_id),
439                directory_id: uuid_from_wire(candidate.candidate_directory_id),
440            })
441            .collect(),
442    })
443}
444
445/// Decodes a Fetch request body (api 1).
446#[must_use]
447pub fn decode_fetch(buf: &[u8]) -> Option<PeerRequest> {
448    let req = decode_request::<FetchRequest>(buf, FETCH_VERSION)?;
449    let from = node_from_wire(req.replica_state.replica_id);
450    let p = req.topics.first()?.partitions.first()?;
451    let replica_directory_id = uuid_from_wire(p.replica_directory_id);
452    Some(PeerRequest::Fetch {
453        cluster_id: request_cluster_id(req.cluster_id.as_deref()).ok()?,
454        from,
455        max_wait_ms: req.max_wait_ms,
456        current_leader_epoch: p.current_leader_epoch,
457        fetch_epoch: epoch_from_wire(p.last_fetched_epoch),
458        fetch_offset: p.fetch_offset,
459        replica_directory_id,
460        high_watermark: p.high_watermark,
461    })
462}
463
464/// The byte window of a snapshot that a `FetchSnapshot` request asks for.
465struct ChunkWindow {
466    position: i64,
467    max_bytes: i32,
468}
469
470/// Encodes a `FetchSnapshot` request body (api 59), as Kafka's
471/// `RaftUtil.singletonFetchSnapshotRequest` builds it.
472fn encode_fetch_snapshot_request(
473    cluster_id: Option<uuid::Uuid>,
474    from: NodeId,
475    current_leader_epoch: i32,
476    snapshot_id: SnapshotId,
477    ChunkWindow {
478        position,
479        max_bytes,
480    }: ChunkWindow,
481) -> Bytes {
482    let (end_offset, epoch) = snapshot_id;
483    let req = FetchSnapshotRequest {
484        cluster_id: cluster_id_to_wire(cluster_id),
485        replica_id: node_to_wire(from),
486        max_bytes,
487        topics: vec![fs_req::TopicSnapshot {
488            name: METADATA_TOPIC.to_owned(),
489            partitions: vec![fs_req::PartitionSnapshot {
490                partition: METADATA_PARTITION,
491                current_leader_epoch,
492                snapshot_id: fs_req::SnapshotId {
493                    end_offset,
494                    epoch,
495                    ..Default::default()
496                },
497                position,
498                ..Default::default()
499            }],
500            ..Default::default()
501        }],
502        ..Default::default()
503    };
504    encode_body(&req, FETCH_SNAPSHOT_VERSION)
505}
506
507/// Decodes a `FetchSnapshot` request body (api 59).
508#[must_use]
509pub fn decode_fetch_snapshot(buf: &[u8]) -> Option<PeerRequest> {
510    let req = decode_request::<FetchSnapshotRequest>(buf, FETCH_SNAPSHOT_VERSION)?;
511    let p = req.topics.first()?.partitions.first()?;
512    Some(PeerRequest::FetchSnapshot {
513        cluster_id: request_cluster_id(req.cluster_id.as_deref()).ok()?,
514        from: node_from_wire(req.replica_id),
515        current_leader_epoch: p.current_leader_epoch,
516        snapshot_id: (p.snapshot_id.end_offset, p.snapshot_id.epoch),
517        position: p.position,
518        max_bytes: req.max_bytes,
519    })
520}
krabka_raft/kraft/transport/wire/request.rs

krabka_raft/kraft/transport/wire/
request.rs
