krabka_raft/network/
dialer.rs

1//! The outbound dial seam the controller hands to the peer sender, and the
2//! PLAINTEXT dialer it falls back to.
3//!
4//! Terminating TLS and SASL on a controller-to-controller dial needs the
5//! broker's `InterBrokerClient`, which this crate cannot depend on, so the dial
6//! is a trait the broker injects an implementation of. `PlaintextDialer` is
7//! what the controller uses when nothing is injected.
8
9use {
10    crate::kraft::types::NodeId,
11    async_trait::async_trait,
12    krabka_client_core::{ClientError, Connection, ConnectionOptions, transport},
13    std::{
14        io::{Error, ErrorKind},
15        net::SocketAddr,
16    },
17};
18
19/// Outbound dialer the controller hands to the peer sender.
20///
21/// `krabka-raft` cannot depend on `krabka-broker`, because that would be a
22/// cycle. The broker therefore supplies an impl that wraps its
23/// `InterBrokerClient`, with TLS and SASL, and injects it through
24/// [`ControllerConfig::dialer`](crate::ControllerConfig). When no dialer is
25/// injected, the controller falls back to a plain `Connection::connect(addr)`,
26/// which is the PLAINTEXT path.
27#[async_trait]
28pub trait OutboundDialer: Send + Sync {
29    /// Opens a `Connection` to the raft peer `target`, which is reachable on
30    /// `addr`. The returned connection has already negotiated `ApiVersions`,
31    /// and `raw_request` can use it immediately.
32    async fn dial(
33        &self,
34        target: NodeId,
35        addr: &str,
36        options: ConnectionOptions,
37    ) -> Result<Connection, ClientError>;
38}
39
40/// Default no-op dialer: it opens a raw `TcpStream` with `Connection::connect`.
41///
42/// The controller uses this dialer when the broker has injected no
43/// `InterBrokerClient`-backed dialer. This is the legacy PLAINTEXT path.
44pub struct PlaintextDialer;
45
46#[async_trait]
47impl OutboundDialer for PlaintextDialer {
48    #[tracing::instrument(level = "debug", skip_all, fields(target = _target.0, addr), err)]
49    async fn dial(
50        &self,
51        _target: NodeId,
52        addr: &str,
53        options: ConnectionOptions,
54    ) -> Result<Connection, ClientError> {
55        // Re-resolve `addr` (a `<host>:<port>`) on every dial. A `StatefulSet`
56        // peer that restarts keeps its stable DNS name but gets a fresh pod IP;
57        // resolving here (rather than once at startup) reaches the new IP.
58        // A literal `ip:port` resolves to itself with no lookup, which is
59        // what keeps this usable on WASI, where a host name is an error.
60        let sock: SocketAddr = transport::resolve(addr)
61            .await
62            .map_err(ClientError::Io)?
63            .into_iter()
64            .next()
65            .ok_or_else(|| {
66                ClientError::Io(Error::new(
67                    ErrorKind::NotFound,
68                    format!("raft peer address {addr:?} resolved to no addresses"),
69                ))
70            })?;
71        Connection::connect(sock, options).await
72    }
73}
krabka_raft/network/dialer.rs

krabka_raft/network/
dialer.rs
