pb_mapper_client/
relay.rs

1//! Relay-carrier configuration and on-demand observations, independent of P2P.
2use pb_mapper_core::{checksum::Credential, config::ResolvedAddrs};
3pub use pb_mapper_protocol::transport::RelayTransport;
4use pb_mapper_protocol::{
5    command::{MessageSerializer, PbConnRequest, PbConnResponse, PbErrorResponse, QuicEndpoint},
6    secure::ClientHeaderSession,
7    transport::CONNECT_TIMEOUT,
8};
9use serde::Serialize;
10use std::{io, net::SocketAddr};
11
12#[cfg(feature = "relay-quic")]
13pub(crate) use quic::CarrierId;
14
15/// Without QUIC every relay stream is a TCP connection of its own, with no
16/// shared carrier to blame for a failure: an attempt's carrier is always
17/// `None`, and charging it compiles away.
18#[cfg(not(feature = "relay-quic"))]
19#[derive(Clone, Copy, Debug)]
20pub(crate) enum CarrierId {}
21
22/// Carrier setup can already receive an authenticated refusal, before a
23/// registration, subscription or administrator request has been sent.
24#[derive(Debug, snafu::Snafu)]
25pub(crate) enum ConnectError {
26    #[snafu(display("{source}"))]
27    Transport { source: std::io::Error },
28    #[snafu(display("{}: {}", response.code, response.message))]
29    Rejected { response: PbErrorResponse },
30}
31
32impl From<std::io::Error> for ConnectError {
33    fn from(source: std::io::Error) -> Self {
34        Self::Transport { source }
35    }
36}
37
38impl From<ConnectError> for std::io::Error {
39    fn from(error: ConnectError) -> Self {
40        match error {
41            ConnectError::Transport { source } => source,
42            error @ ConnectError::Rejected { .. } => {
43                Self::new(std::io::ErrorKind::PermissionDenied, error)
44            }
45        }
46    }
47}
48
49/// The refusal code for a relay that offers no QUIC: one started without a
50/// QUIC listener, or one from before QUIC, which cannot read the question.
51/// Final, since retrying the same relay cannot change it, and so a tunnel
52/// asked to use QUIC there fails rather than retrying forever.
53pub const QUIC_UNAVAILABLE: &str = "quic_unavailable";
54
55/// What an older relay answers a request it cannot decode, which is how it
56/// answers [`PbConnRequest::QuicEndpoint`].
57const REQUEST_DECODE_FAILED: &str = "request_decode_failed";
58
59fn quic_unavailable() -> ConnectError {
60    ConnectError::Rejected {
61        response: PbErrorResponse {
62            code: QUIC_UNAVAILABLE.into(),
63            message: "the relay does not offer QUIC; select TCP to use it".into(),
64            retryable: false,
65            server_time: 0,
66        },
67    }
68}
69
70/// Ask the relay at `addr`, over one authenticated TCP exchange, where it
71/// takes QUIC. Answers the relay address that TCP reached, whose IP QUIC
72/// shares, with the advertised endpoint.
73pub(crate) async fn discover(
74    addr: SocketAddr,
75    credential: Credential,
76) -> Result<(SocketAddr, QuicEndpoint), ConnectError> {
77    let mut tcp = tokio::net::TcpStream::connect(addr).await?;
78    tcp.set_nodelay(true)?;
79    let relay_addr = tcp.peer_addr()?;
80    let session = ClientHeaderSession::new_v2(&credential).map_err(io::Error::other)?;
81    let response = session
82        .exchange(
83            &mut tcp,
84            &PbConnRequest::QuicEndpoint
85                .encode()
86                .map_err(io::Error::other)?,
87            CONNECT_TIMEOUT,
88        )
89        .await
90        .map_err(io::Error::other)?;
91    match PbConnResponse::decode(&response).map_err(io::Error::other)? {
92        PbConnResponse::QuicEndpoint(Some(endpoint))
93            if endpoint.port != 0
94                && !endpoint.certificate.is_empty()
95                && endpoint.certificate.len() <= 4096 =>
96        {
97            Ok((relay_addr, endpoint))
98        }
99        // Named apart from an unexpected answer: a misconfigured relay is
100        // found by what it advertised.
101        PbConnResponse::QuicEndpoint(Some(endpoint)) => Err(io::Error::new(
102            io::ErrorKind::InvalidData,
103            format!(
104                "the relay advertised an unusable QUIC endpoint: UDP port {}, \
105                 a {}-byte certificate",
106                endpoint.port,
107                endpoint.certificate.len()
108            ),
109        )
110        .into()),
111        PbConnResponse::QuicEndpoint(None) => Err(quic_unavailable()),
112        PbConnResponse::Error(error) if error.code == REQUEST_DECODE_FAILED => {
113            Err(quic_unavailable())
114        }
115        PbConnResponse::Error(response) => Err(ConnectError::Rejected { response }),
116        other => Err(io::Error::new(
117            io::ErrorKind::InvalidData,
118            format!("unexpected QUIC discovery response: {other:?}"),
119        )
120        .into()),
121    }
122}
123
124/// The UDP port the relay offers QUIC on, or `None` when it offers none.
125pub(crate) async fn offered_quic_port(
126    addrs: &ResolvedAddrs,
127    credential: Credential,
128) -> Result<Option<u16>, ConnectError> {
129    match crate::addr::race_candidates(addrs.as_slice(), CONNECT_TIMEOUT, move |addr| {
130        discover(addr, credential)
131    })
132    .await
133    {
134        Ok((_, endpoint)) => Ok(Some(endpoint.port)),
135        Err(ConnectError::Rejected { response }) if response.code == QUIC_UNAVAILABLE => Ok(None),
136        Err(error) => Err(error),
137    }
138}
139
140#[cfg(feature = "relay-quic")]
141pub(crate) mod quic;
142
143/// The relay carrier as this client's pool sees it, sampled on request.
144#[derive(Clone, Debug, Serialize)]
145#[non_exhaustive]
146pub struct RelayDiagnostics {
147    pub transport: RelayTransport,
148    /// Successful QUIC handshakes over this pool's lifetime.
149    pub connections_created: u64,
150    pub streams_opened: u64,
151    pub streams_reused: u64,
152    pub last_error: Option<String>,
153    /// Every open QUIC connection: those the pool hands out, then those it
154    /// has retired that streams opened earlier still carry traffic on.
155    pub connections: Vec<QuicConnectionDiagnostics>,
156}
157impl RelayDiagnostics {
158    pub(crate) fn empty(transport: RelayTransport) -> Self {
159        Self {
160            transport,
161            connections_created: 0,
162            streams_opened: 0,
163            streams_reused: 0,
164            last_error: None,
165            connections: Vec::new(),
166        }
167    }
168}
169
170/// Cumulative Quinn counters for one live connection; sampled only on request.
171#[derive(Clone, Debug, Serialize)]
172#[non_exhaustive]
173pub struct QuicConnectionDiagnostics {
174    pub connection_id: u64,
175    pub peer_addr: std::net::SocketAddr,
176    /// Whether the pool still hands it out. A retired one stays open only
177    /// for the streams already on it.
178    pub pooled: bool,
179    pub rtt_us: u64,
180    pub congestion_window: u64,
181    pub mtu: u16,
182    pub sent_packets: u64,
183    pub lost_packets: u64,
184    pub congestion_events: u64,
185    pub sent_bytes: u64,
186    pub received_bytes: u64,
187}
pb_mapper_client/relay.rs

pb_mapper_client/
relay.rs
