krabka_raft/kraft/transport/wire/
response.rs

1//! The `PeerResponse` body codec.
2//!
3//! `PeerResponse` is the flat response vocabulary the engine reasons in. This
4//! module maps each variant onto the generated KIP-595 response message at its
5//! captured version, and decodes each one back into the variant the sending
6//! engine feeds to the core.
7
8use {
9    super::codec::{
10        FETCH_SNAPSHOT_VERSION, FETCH_VERSION, METADATA_PARTITION, METADATA_TOPIC,
11        METADATA_TOPIC_ID, QUORUM_EPOCH_VERSION, VOTE_VERSION, encode_body, epoch_from_wire,
12        epoch_to_wire, node_from_wire, node_to_wire, records_payload_to_bytes,
13    },
14    crate::kraft::{
15        snapshot_fetch::SnapshotId,
16        types::{Epoch, LogOffsetMetadata, NodeId},
17    },
18    bytes::Bytes,
19    krabka_protocol::{
20        Decode,
21        owned::{
22            begin_quorum_epoch_response::{self as bqe_resp, BeginQuorumEpochResponse, TopicData},
23            end_quorum_epoch_response::{self as eqe_resp, EndQuorumEpochResponse},
24            fetch_response::{self as fetch_resp, FetchResponse},
25            fetch_snapshot_response::{self as fs_resp, FetchSnapshotResponse},
26            vote_response::{self as vote_resp, VoteResponse},
27        },
28        records::RecordsPayload,
29    },
30};
31
32#[cfg(test)]
33mod tests;
34
35/// A peer RPC response body. The sending engine decodes it back into the
36/// matching `Receive*Response` event, or applies it directly for Fetch.
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub enum PeerResponse {
39    /// A Vote answer: the partition error, the grant, and the responder's
40    /// view of the leader and epoch, which Kafka's `handleVoteResponse` hands
41    /// to `maybeHandleCommonResponse`.
42    Vote {
43        error_code: i16,
44        granted: bool,
45        leader: QuorumLeader,
46    },
47    /// `BeginQuorumEpoch` and `EndQuorumEpoch` acks carry the responder's
48    /// epoch. They produce no core event.
49    Ack {
50        epoch: Epoch,
51    },
52    Fetch(FetchAnswer),
53    FetchSnapshot {
54        snapshot_id: SnapshotId,
55        size: i64,
56        position: i64,
57        bytes: Bytes,
58        error_code: i16,
59    },
60}
61
62/// A Fetch answer, in the one shape Kafka's
63/// `KafkaRaftClient.buildFetchResponse` gives both a served fetch and a
64/// refused one.
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub struct FetchAnswer {
67    /// The partition `ErrorCode`: `NONE` from the leader, or
68    /// `validateLeaderOnlyRequest`'s `FENCED_LEADER_EPOCH`,
69    /// `UNKNOWN_LEADER_EPOCH` or `NOT_LEADER_OR_FOLLOWER`.
70    pub error_code: i16,
71    /// `CurrentLeader`, and the leader's `NodeEndpoints` entry.
72    pub leader: QuorumLeader,
73    pub diverging: Option<LogOffsetMetadata>,
74    /// When set, the follower's fetch offset is below the leader's pruned
75    /// log-start, and the follower must `FetchSnapshot` this snapshot
76    /// instead.
77    pub snapshot_id: Option<SnapshotId>,
78    /// Leader's high watermark at serve time, -1 in a refusal.
79    pub hwm: i64,
80    /// The responder's log start offset.
81    pub log_start_offset: i64,
82    /// Verbatim concatenated `RecordBatch` bytes for `[fetch_offset, log_end)`.
83    pub records: Bytes,
84}
85
86impl FetchAnswer {
87    /// Encodes the answer as a Fetch response body (api 1) at `version`, the
88    /// version of the request it answers.
89    #[must_use]
90    pub fn encode(&self, version: i16) -> Bytes {
91        // Kafka always sets `Records`, to `MemoryRecords.EMPTY` when
92        // there are none, and leaves `AbortedTransactions` at its
93        // generated default, an empty list: neither is null on the
94        // wire.
95        let mut partition = fetch_resp::PartitionData {
96            partition_index: METADATA_PARTITION,
97            error_code: self.error_code,
98            high_watermark: self.hwm,
99            log_start_offset: self.log_start_offset,
100            aborted_transactions: Some(Vec::new()),
101            current_leader: fetch_resp::LeaderIdAndEpoch {
102                leader_id: self.leader.leader_id_to_wire(),
103                leader_epoch: epoch_to_wire(self.leader.epoch),
104                ..Default::default()
105            },
106            records: Some(RecordsPayload::Raw(self.records.clone())),
107            ..Default::default()
108        };
109        if let Some(point) = self.diverging {
110            partition.diverging_epoch = fetch_resp::EpochEndOffset {
111                epoch: epoch_to_wire(point.epoch),
112                end_offset: point.offset,
113                ..Default::default()
114            };
115        }
116        if let Some((end_offset, epoch)) = self.snapshot_id {
117            partition.snapshot_id = fetch_resp::SnapshotId {
118                end_offset,
119                epoch,
120                ..Default::default()
121            };
122        }
123        // `RaftUtil.singletonFetchResponse`: from v16 `NodeEndpoints`
124        // names the leader when its id and endpoint are known.
125        let resp = FetchResponse {
126            responses: vec![fetch_resp::FetchableTopicResponse {
127                topic: METADATA_TOPIC.to_owned(),
128                topic_id: METADATA_TOPIC_ID,
129                partitions: vec![partition],
130                ..Default::default()
131            }],
132            node_endpoints: self
133                .leader
134                .node_endpoint()
135                .filter(|_| version >= 17)
136                .map(
137                    |NodeEndpointEntry {
138                         node_id,
139                         host,
140                         port,
141                     }| fetch_resp::NodeEndpoint {
142                        node_id,
143                        host,
144                        port: i32::from(port),
145                        ..Default::default()
146                    },
147                )
148                .into_iter()
149                .collect(),
150            ..Default::default()
151        };
152        encode_body(&resp, version)
153    }
154}
155
156/// Encodes Kafka's bare top-level Fetch error response at `version`:
157/// `new FetchResponseData().setErrorCode(error)`, with no topic.
158#[must_use]
159pub fn encode_fetch_top_level_error(error_code: i16, version: i16) -> Bytes {
160    let resp = FetchResponse {
161        error_code,
162        ..Default::default()
163    };
164    encode_body(&resp, version)
165}
166
167/// What a quorum RPC response says about the responder's view of the leader,
168/// as Kafka's `RaftUtil.singleton*Response` helpers fill it.
169#[derive(Debug, Clone, PartialEq, Eq)]
170pub struct QuorumLeader {
171    /// The leader of the epoch, or `None` (wire -1) while it is unknown.
172    pub leader_id: Option<NodeId>,
173    /// The responder's epoch.
174    pub epoch: Epoch,
175    /// The leader's controller-listener host and port. The response carries it
176    /// in `NodeEndpoints` only when the leader is known.
177    pub endpoint: Option<LeaderEndpoint>,
178}
179
180/// The host and port a leader is reached at.
181#[derive(Debug, Clone, PartialEq, Eq)]
182pub struct LeaderEndpoint {
183    /// The host name or address.
184    pub host: String,
185    /// The port.
186    pub port: u16,
187}
188
189/// A `NodeEndpoints` entry of a quorum response, before it takes the shape of
190/// a particular response's generated type.
191struct NodeEndpointEntry {
192    node_id: i32,
193    host: String,
194    port: u16,
195}
196
197impl QuorumLeader {
198    fn leader_id_to_wire(&self) -> i32 { self.leader_id.map_or(-1, node_to_wire) }
199
200    /// The leader a response names on the wire. Its endpoint is the
201    /// `NodeEndpoints` entry, given as `(node id, host, port)`, whose node id
202    /// is the leader's, as Kafka's `Endpoints.fromFetchResponse` and
203    /// `Endpoints.fromVoteResponse` select it; an entry for any other node is
204    /// ignored.
205    fn from_wire<'a>(
206        wire_leader: i32,
207        wire_epoch: i32,
208        endpoints: impl IntoIterator<Item = (i32, &'a str, i32)>,
209    ) -> Self {
210        let leader_id = (wire_leader >= 0).then(|| node_from_wire(wire_leader));
211        let endpoint = leader_id.and_then(|_| {
212            let (_, host, port) = endpoints
213                .into_iter()
214                .find(|(node_id, ..)| *node_id == wire_leader)?;
215            u16::try_from(port).ok().map(|port| LeaderEndpoint {
216                host: host.to_owned(),
217                port,
218            })
219        });
220        Self {
221            leader_id,
222            epoch: epoch_from_wire(wire_epoch),
223            endpoint,
224        }
225    }
226
227    /// The `NodeEndpoints` entry, when the leader and its endpoint are known.
228    fn node_endpoint(&self) -> Option<NodeEndpointEntry> {
229        let leader_id = self.leader_id?;
230        let LeaderEndpoint { host, port } = self.endpoint.clone()?;
231        Some(NodeEndpointEntry {
232            node_id: node_to_wire(leader_id),
233            host,
234            port,
235        })
236    }
237}
238
239/// Encodes a Vote response body (api 52) at `version`, the version of the
240/// request it answers.
241///
242/// A `top_level_error` other than 0 is Kafka's bare error response: the code
243/// and nothing else. Otherwise the body names the metadata partition with
244/// `partition_error`, the grant and the responder's leader view.
245#[must_use]
246pub fn encode_vote_response(
247    top_level_error: i16,
248    partition_error: i16,
249    vote_granted: bool,
250    leader: &QuorumLeader,
251    version: i16,
252) -> Bytes {
253    let resp = if top_level_error == 0 {
254        VoteResponse {
255            topics: vec![vote_resp::TopicData {
256                topic_name: METADATA_TOPIC.to_owned(),
257                partitions: vec![vote_resp::PartitionData {
258                    partition_index: METADATA_PARTITION,
259                    error_code: partition_error,
260                    leader_id: leader.leader_id_to_wire(),
261                    leader_epoch: epoch_to_wire(leader.epoch),
262                    vote_granted,
263                    ..Default::default()
264                }],
265                ..Default::default()
266            }],
267            node_endpoints: leader
268                .node_endpoint()
269                .map(
270                    |NodeEndpointEntry {
271                         node_id,
272                         host,
273                         port,
274                     }| vote_resp::NodeEndpoint {
275                        node_id,
276                        host,
277                        port,
278                        ..Default::default()
279                    },
280                )
281                .into_iter()
282                .collect(),
283            ..Default::default()
284        }
285    } else {
286        VoteResponse {
287            error_code: top_level_error,
288            ..Default::default()
289        }
290    };
291    encode_body(&resp, version)
292}
293
294// Begin/EndQuorumEpoch share Kafka's complete response layout.
295krabka_macros::quorum_epoch_encoder! {
296    /// Encodes a `BeginQuorumEpoch` response body (api 53) at `version`, with the
297    /// same shape rules as [`encode_vote_response`].
298    #[must_use]
299    pub fn encode_begin_quorum_epoch_response(
300        top_level_error: i16,
301        partition_error: i16,
302        leader: &QuorumLeader,
303        version: i16,
304    ) -> Bytes
305    [response = BeginQuorumEpochResponse, wire = bqe_resp]
306    {}
307}
308krabka_macros::quorum_epoch_encoder! {
309    /// Encodes an `EndQuorumEpoch` response body (api 54) at `version`, with the
310    /// same shape rules as [`encode_vote_response`].
311    #[must_use]
312    pub fn encode_end_quorum_epoch_response(
313        top_level_error: i16,
314        partition_error: i16,
315        leader: &QuorumLeader,
316        version: i16,
317    ) -> Bytes
318    [response = EndQuorumEpochResponse, wire = eqe_resp]
319    {}
320}
321
322/// The one partition of a `FetchSnapshot` answer that names a partition.
323#[derive(Debug, Clone, PartialEq, Eq)]
324pub struct FetchSnapshotPartition {
325    /// The topic the request named. It is `__cluster_metadata` except in the
326    /// `UNKNOWN_TOPIC_OR_PARTITION` answer.
327    pub topic: String,
328    /// The partition the request named.
329    pub index: i32,
330    pub error_code: i16,
331    /// Whether `CurrentLeader` carries the responder's leader view. Kafka's
332    /// `addQuorumLeader` fills it in every answer except
333    /// `UNKNOWN_TOPIC_OR_PARTITION`.
334    pub current_leader: bool,
335    /// The chunk served, when one is.
336    pub chunk: Option<SnapshotChunk>,
337}
338
339/// A served chunk of a snapshot.
340#[derive(Debug, Clone, PartialEq, Eq)]
341pub struct SnapshotChunk {
342    /// The snapshot the chunk belongs to.
343    pub snapshot_id: SnapshotId,
344    /// The size of the whole snapshot.
345    pub size: i64,
346    /// The byte position of the chunk within the snapshot.
347    pub position: i64,
348    /// The chunk bytes.
349    pub bytes: Bytes,
350}
351
352/// Encodes a `FetchSnapshot` response body (api 59) at `version`, as Kafka's
353/// `RaftUtil.singletonFetchSnapshotResponse` builds it.
354///
355/// `partition` `None` is Kafka's bare top-level error response,
356/// `FetchSnapshotResponse.withTopLevelError`. Otherwise `NodeEndpoints` names
357/// the leader when the leader and its endpoint are known.
358#[must_use]
359pub fn encode_fetch_snapshot_answer(
360    top_level_error: i16,
361    partition: Option<FetchSnapshotPartition>,
362    leader: &QuorumLeader,
363    version: i16,
364) -> Bytes {
365    let Some(partition) = partition else {
366        let resp = FetchSnapshotResponse {
367            error_code: top_level_error,
368            ..Default::default()
369        };
370        return encode_body(&resp, version);
371    };
372    let mut snapshot = fs_resp::PartitionSnapshot {
373        index: partition.index,
374        error_code: partition.error_code,
375        ..Default::default()
376    };
377    if partition.current_leader {
378        snapshot.current_leader = fs_resp::LeaderIdAndEpoch {
379            leader_id: leader.leader_id_to_wire(),
380            leader_epoch: epoch_to_wire(leader.epoch),
381            ..Default::default()
382        };
383    }
384    if let Some(SnapshotChunk {
385        snapshot_id: (end_offset, epoch),
386        size,
387        position,
388        bytes,
389    }) = partition.chunk
390    {
391        snapshot.snapshot_id = fs_resp::SnapshotId {
392            end_offset,
393            epoch,
394            ..Default::default()
395        };
396        snapshot.size = size;
397        snapshot.position = position;
398        snapshot.unaligned_records = RecordsPayload::Raw(bytes);
399    }
400    let resp = FetchSnapshotResponse {
401        error_code: top_level_error,
402        topics: vec![fs_resp::TopicSnapshot {
403            name: partition.topic,
404            partitions: vec![snapshot],
405            ..Default::default()
406        }],
407        node_endpoints: leader
408            .node_endpoint()
409            .map(
410                |NodeEndpointEntry {
411                     node_id,
412                     host,
413                     port,
414                 }| fs_resp::NodeEndpoint {
415                    node_id,
416                    host,
417                    port,
418                    ..Default::default()
419                },
420            )
421            .into_iter()
422            .collect(),
423        ..Default::default()
424    };
425    encode_body(&resp, version)
426}
427
428/// Encodes a `FetchSnapshot` response body (api 59).
429fn encode_fetch_snapshot_response(
430    snapshot_id: SnapshotId,
431    size: i64,
432    position: i64,
433    bytes: &Bytes,
434    error_code: i16,
435) -> Bytes {
436    let (end_offset, epoch) = snapshot_id;
437    let resp = FetchSnapshotResponse {
438        topics: vec![fs_resp::TopicSnapshot {
439            name: METADATA_TOPIC.to_owned(),
440            partitions: vec![fs_resp::PartitionSnapshot {
441                index: METADATA_PARTITION,
442                error_code,
443                snapshot_id: fs_resp::SnapshotId {
444                    end_offset,
445                    epoch,
446                    ..Default::default()
447                },
448                size,
449                position,
450                unaligned_records: RecordsPayload::Raw(bytes.clone()),
451                ..Default::default()
452            }],
453            ..Default::default()
454        }],
455        ..Default::default()
456    };
457    encode_body(&resp, FETCH_SNAPSHOT_VERSION)
458}
459
460impl PeerResponse {
461    #[must_use]
462    pub fn encode(&self) -> Bytes {
463        match self {
464            Self::Vote {
465                error_code,
466                granted,
467                leader,
468            } => encode_vote_response(0, *error_code, *granted, leader, VOTE_VERSION),
469            Self::Ack { epoch } => {
470                // A Begin/End ack is encoded as the corresponding
471                // BeginQuorumEpochResponse with the responder's leader_epoch.
472                let resp = BeginQuorumEpochResponse {
473                    topics: vec![TopicData {
474                        topic_name: METADATA_TOPIC.to_owned(),
475                        partitions: vec![bqe_resp::PartitionData {
476                            partition_index: METADATA_PARTITION,
477                            leader_id: -1,
478                            leader_epoch: epoch_to_wire(*epoch),
479                            ..Default::default()
480                        }],
481                        ..Default::default()
482                    }],
483                    ..Default::default()
484                };
485                encode_body(&resp, QUORUM_EPOCH_VERSION)
486            }
487            Self::Fetch(answer) => answer.encode(FETCH_VERSION),
488            Self::FetchSnapshot {
489                snapshot_id,
490                size,
491                position,
492                bytes,
493                error_code,
494            } => encode_fetch_snapshot_response(*snapshot_id, *size, *position, bytes, *error_code),
495        }
496    }
497
498    /// Decodes a Vote response body (api 52). The round, pre-vote or
499    /// real, is not on the wire. The engine infers it from the candidate's
500    /// role.
501    #[must_use]
502    pub fn decode_vote(buf: &[u8]) -> Option<Self> {
503        let mut cur = buf;
504        let resp = VoteResponse::decode(&mut cur, VOTE_VERSION).ok()?;
505        let p = resp.topics.first()?.partitions.first()?;
506        Some(Self::Vote {
507            error_code: p.error_code,
508            granted: p.vote_granted,
509            leader: QuorumLeader::from_wire(
510                p.leader_id,
511                p.leader_epoch,
512                resp.node_endpoints.iter().map(|endpoint| {
513                    (
514                        endpoint.node_id,
515                        endpoint.host.as_str(),
516                        i32::from(endpoint.port),
517                    )
518                }),
519            ),
520        })
521    }
522
523    /// Decodes a `BeginQuorumEpoch` or `EndQuorumEpoch` ack body
524    /// (api 53 and api 54).
525    #[must_use]
526    pub fn decode_ack(buf: &[u8]) -> Option<Self> {
527        let mut cur = buf;
528        let resp = BeginQuorumEpochResponse::decode(&mut cur, QUORUM_EPOCH_VERSION).ok()?;
529        let p = resp.topics.first()?.partitions.first()?;
530        Some(Self::Ack {
531            epoch: epoch_from_wire(p.leader_epoch),
532        })
533    }
534
535    /// Decodes a Fetch response body (api 1).
536    ///
537    /// The leader's endpoint is the `NodeEndpoints` entry whose node id is
538    /// `CurrentLeader.LeaderId`, as Kafka's `Endpoints.fromFetchResponse`
539    /// selects it; an entry for any other node is ignored.
540    #[must_use]
541    pub fn decode_fetch(buf: &[u8]) -> Option<Self> {
542        let mut cur = buf;
543        let resp = FetchResponse::decode(&mut cur, FETCH_VERSION).ok()?;
544        let p = resp.responses.first()?.partitions.first()?;
545        let leader = QuorumLeader::from_wire(
546            p.current_leader.leader_id,
547            p.current_leader.leader_epoch,
548            resp.node_endpoints
549                .iter()
550                .map(|endpoint| (endpoint.node_id, endpoint.host.as_str(), endpoint.port)),
551        );
552        // diverging_epoch defaults to (-1, -1); a real divergence carries a
553        // non-negative end_offset.
554        let diverging = (p.diverging_epoch.end_offset >= 0).then(|| LogOffsetMetadata {
555            offset: p.diverging_epoch.end_offset,
556            epoch: epoch_from_wire(p.diverging_epoch.epoch),
557        });
558        let snapshot_id = (p.snapshot_id.end_offset >= 0)
559            .then_some((p.snapshot_id.end_offset, p.snapshot_id.epoch));
560        let records = p
561            .records
562            .as_ref()
563            .map_or_else(Bytes::new, records_payload_to_bytes);
564        Some(Self::Fetch(FetchAnswer {
565            error_code: p.error_code,
566            leader,
567            diverging,
568            snapshot_id,
569            hwm: p.high_watermark,
570            log_start_offset: p.log_start_offset,
571            records,
572        }))
573    }
574
575    /// Decodes a `FetchSnapshot` response body (api 59).
576    #[must_use]
577    pub fn decode_fetch_snapshot(buf: &[u8]) -> Option<Self> {
578        let mut cur = buf;
579        let resp = FetchSnapshotResponse::decode(&mut cur, FETCH_SNAPSHOT_VERSION).ok()?;
580        let p = resp.topics.first()?.partitions.first()?;
581        let bytes = records_payload_to_bytes(&p.unaligned_records);
582        Some(Self::FetchSnapshot {
583            snapshot_id: (p.snapshot_id.end_offset, p.snapshot_id.epoch),
584            size: p.size,
585            position: p.position,
586            bytes,
587            error_code: p.error_code,
588        })
589    }
590}
krabka_raft/kraft/transport/wire/response.rs

krabka_raft/kraft/transport/wire/
response.rs
