krabka_raft/kraft/transport/
peer_sender.rs

1//! The outbound peer-RPC seam and its no-op implementation.
2//!
3//! [`PeerSender`] is the trait the engine sends already-encoded request bodies
4//! through. Production backs it with TCP, tests back it in memory, and
5//! [`NullPeerSender`] backs the single-voter case where no peer RPC is ever
6//! sent.
7
8use {
9    crate::{error::RaftError, kraft::types::NodeId},
10    bytes::Bytes,
11};
12
13/// Outbound peer RPC sender.
14///
15/// It encodes nothing itself. The event loop hands it the already-encoded
16/// request body (see [`wire`](super::wire)) and the destination peer. The impl then dials
17/// the peer, sends the body, and returns the raw response body.
18///
19/// This matches the `async_trait` mechanism that
20/// [`OutboundDialer`](crate::network::OutboundDialer) uses.
21#[async_trait::async_trait]
22pub trait PeerSender: Send + Sync {
23    /// Sends `body`, a request for `api_key`, to `peer` and returns the raw
24    /// response body.
25    ///
26    /// # Errors
27    /// Returns [`RaftError`] if the peer is unreachable or the RPC fails.
28    async fn send(&self, peer: NodeId, api_key: i16, body: Bytes) -> Result<Bytes, RaftError>;
29
30    /// Probe an advertised candidate endpoint through the configured outbound
31    /// security dialer and verify its finalized `kraft.version` support.
32    async fn probe_kraft_version(
33        &self,
34        _address: &str,
35        _finalized_version: u16,
36    ) -> Result<bool, RaftError> {
37        Err(RaftError::ChangeRejected(
38            "candidate probing is unavailable on this transport".into(),
39        ))
40    }
41
42    /// Replace the peer endpoint table after applying a `VotersRecord`.
43    fn update_voters(&self, _voters: &krabka_metadata::VoterSet) {}
44
45    /// Transport-only bootstrap peers used by an observer with no voter view.
46    fn discovery_peers(&self) -> Vec<NodeId> { Vec::new() }
47
48    /// Address `leader` at `address`, the `host:port` it announced in a
49    /// response's `NodeEndpoints`. A voter-set listener for the same node
50    /// takes precedence.
51    fn remember_leader_endpoint(&self, _leader: NodeId, _address: String) {}
52}
53
54/// A no-op `PeerSender` for single-voter and no-network tests. Every send fails
55/// as unreachable.
56///
57/// A single voter never sends peer RPCs, because it wins its own election
58/// immediately. This sender therefore lets the contract tests run without a
59/// real transport.
60pub struct NullPeerSender;
61
62#[async_trait::async_trait]
63impl PeerSender for NullPeerSender {
64    async fn send(&self, peer: NodeId, _api_key: i16, _body: Bytes) -> Result<Bytes, RaftError> {
65        Err(RaftError::NotLeader {
66            current_leader: Some(peer),
67        })
68    }
69}
70
71#[cfg(test)]
72mod tests {
73    use {super::*, crate::kraft::transport::api_key, assert2::assert};
74
75    #[tokio::test]
76    async fn null_peer_sender_reports_target_as_current_leader() {
77        let err = NullPeerSender
78            .send(NodeId(7), api_key::FETCH, Bytes::new())
79            .await
80            .expect_err("null sender should reject peer sends");
81
82        assert2::assert!(matches!(err, RaftError::NotLeader {
83            current_leader: Some(NodeId(7))
84        }));
85    }
86
87    #[tokio::test]
88    async fn null_peer_sender_rejects_candidate_probing_and_has_empty_discovery() {
89        let res = NullPeerSender
90            .probe_kraft_version("127.0.0.1:9093", 1)
91            .await;
92        assert2::assert!(matches!(res, Err(RaftError::ChangeRejected(_))));
93        assert2::assert!(NullPeerSender.discovery_peers().is_empty());
94    }
95}
krabka_raft/kraft/transport/peer_sender.rs

krabka_raft/kraft/transport/
peer_sender.rs
