krabka_raft/kraft/transport/
wire.rs

1//! Real KIP-595 peer-RPC body codec.
2//!
3//! The engine's loop reasons in terms of the flat `PeerRequest` and
4//! `PeerResponse` enums. This module maps each variant to and from the genuine
5//! generated KIP-595 message bodies. Those bodies are header-less, because the
6//! framing layer in `server.rs` and `network.rs` adds the request header and
7//! the response header. The captured wire versions are Vote v2,
8//! `BeginQuorumEpoch` v1, `EndQuorumEpoch` v1, and Fetch v18. Krabka-to-Krabka
9//! replication rides these exact bytes.
10//!
11//! The metadata log is the single `KRaft` topic `__cluster_metadata`, partition
12//! 0, so every RPC body carries exactly one topic and exactly one partition.
13//! Kafka's `VoteResponse` carries no pre-vote field. A candidate matches a
14//! reply to its round from its own `Prospective` or `Candidate` role, so Krabka
15//! encodes a byte-faithful `VoteResponse` and the core infers the round itself
16//! (KIP-996).
17//!
18//! The shared constants and integer conversions live in `codec`, the request
19//! bodies in `request`, and the response bodies in `response`.
20
21mod codec;
22mod request;
23mod response;
24
25/// The versions [`PeerRequest`] encodes each peer RPC at. Public because a
26/// broker-only observer sends `FetchSnapshot` itself, and the version on its
27/// request header has to be the one the body was encoded at.
28pub use self::codec::{FETCH_SNAPSHOT_VERSION, FETCH_VERSION, QUORUM_EPOCH_VERSION, VOTE_VERSION};
29pub(crate) use self::{
30    codec::{
31        METADATA_PARTITION, METADATA_TOPIC, METADATA_TOPIC_ID, epoch_from_wire, node_from_wire,
32        uuid_from_wire,
33    },
34    request::{decode_request, parse_cluster_id},
35};
36pub use self::{
37    request::{
38        PeerRequest, PreferredCandidate, decode_begin, decode_end, decode_fetch,
39        decode_fetch_snapshot, decode_vote, fetch_replica_id,
40    },
41    response::{
42        FetchAnswer, FetchSnapshotPartition, LeaderEndpoint, PeerResponse, QuorumLeader,
43        SnapshotChunk, encode_begin_quorum_epoch_response, encode_end_quorum_epoch_response,
44        encode_fetch_snapshot_answer, encode_fetch_top_level_error, encode_vote_response,
45    },
46};
47
48#[cfg(test)]
49mod tests {
50    use {
51        super::{
52            PeerRequest, PeerResponse,
53            codec::{FETCH_VERSION, METADATA_TOPIC_ID},
54        },
55        crate::kraft::types::NodeId,
56        assert2::assert,
57        bytes::Bytes,
58    };
59
60    #[test]
61    fn fetch_wire_carries_metadata_topic_id() {
62        use krabka_protocol::{
63            Decode,
64            owned::{fetch_request::FetchRequest, fetch_response::FetchResponse},
65        };
66        let req = PeerRequest::Fetch {
67            from: NodeId(2),
68            current_leader_epoch: 1,
69            fetch_epoch: 1,
70            fetch_offset: 5,
71            replica_directory_id: uuid::Uuid::nil(),
72            cluster_id: None,
73            max_wait_ms: 0,
74            high_watermark: -1,
75        };
76        let mut c = &req.encode()[..];
77        let dreq = FetchRequest::decode(&mut c, FETCH_VERSION).unwrap();
78        assert2::assert!(dreq.topics[0].topic_id == METADATA_TOPIC_ID);
79
80        let resp = PeerResponse::Fetch(super::FetchAnswer {
81            error_code: 0,
82            leader: super::QuorumLeader {
83                leader_id: Some(NodeId(1)),
84                epoch: 4,
85                endpoint: None,
86            },
87            diverging: None,
88            snapshot_id: None,
89            hwm: 0,
90            log_start_offset: 0,
91            records: Bytes::new(),
92        });
93        let mut c2 = &resp.encode()[..];
94        let dresp = FetchResponse::decode(&mut c2, FETCH_VERSION).unwrap();
95        assert2::assert!(dresp.responses[0].topic_id == METADATA_TOPIC_ID);
96    }
97}
krabka_raft/kraft/transport/wire.rs

krabka_raft/kraft/transport/
wire.rs
