relay_knowledge/net/http/
outbound.rs

1use std::{error::Error, fmt, io, time::Duration};
2
3use serde_json::Value;
4use tokio::io::{AsyncReadExt, AsyncWriteExt};
5
6use crate::net::qos::{QosPolicy, QosRuntime, RejectReason};
7
8use super::{HttpConfig, qos_request_context_active};
9
10/// Builds an async outbound JSON client from validated network policy.
11pub fn outbound_json_client(config: &HttpConfig) -> Result<reqwest::Client, OutboundClientError> {
12    outbound_json_client_with_policy(config, None, None)
13}
14
15/// Builds an async outbound JSON client with request-scoped transport policy.
16pub fn outbound_json_client_with_policy(
17    config: &HttpConfig,
18    ssl_verify: Option<bool>,
19    connect_timeout: Option<Duration>,
20) -> Result<reqwest::Client, OutboundClientError> {
21    let mut builder = reqwest::Client::builder()
22        .timeout(config.request_timeout)
23        .danger_accept_invalid_certs(!ssl_verify.unwrap_or(config.proxy.ssl_verify));
24    if let Some(timeout) = connect_timeout {
25        builder = builder.connect_timeout(timeout);
26    }
27    if let Some(proxy_url) = &config.proxy.proxy {
28        let no_proxy = reqwest::NoProxy::from_string(&config.proxy.no_proxy_rules.join(","));
29        let proxy = reqwest::Proxy::all(proxy_url)
30            .map_err(|error| OutboundClientError {
31                message: error.to_string(),
32            })?
33            .no_proxy(no_proxy);
34        builder = builder.proxy(proxy);
35    }
36
37    builder.build().map_err(|error| OutboundClientError {
38        message: error.to_string(),
39    })
40}
41
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct OutboundClientError {
44    pub message: String,
45}
46impl fmt::Display for OutboundClientError {
47    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
48        self.message.fmt(formatter)
49    }
50}
51
52impl Error for OutboundClientError {}
53
54/// Error raised by bounded outbound JSON HTTP calls.
55#[derive(Debug)]
56pub enum HttpClientError {
57    InvalidUrl(String),
58    QosRejected(RejectReason),
59    Io(io::Error),
60    Timeout,
61    InvalidResponse,
62    ResponseStatus(u16),
63    ResponseJson(serde_json::Error),
64}
65
66impl fmt::Display for HttpClientError {
67    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
68        match self {
69            Self::InvalidUrl(value) => write!(formatter, "invalid HTTP worker URL: {value}"),
70            Self::QosRejected(reason) => write!(
71                formatter,
72                "HTTP worker request rejected by QoS: {}",
73                reason.as_str()
74            ),
75            Self::Io(error) => write!(formatter, "HTTP worker request failed: {error}"),
76            Self::Timeout => write!(formatter, "HTTP worker request timed out"),
77            Self::InvalidResponse => write!(formatter, "HTTP worker returned invalid response"),
78            Self::ResponseStatus(status) => {
79                write!(formatter, "HTTP worker returned status {status}")
80            }
81            Self::ResponseJson(error) => {
82                write!(formatter, "HTTP worker returned invalid JSON: {error}")
83            }
84        }
85    }
86}
87
88impl Error for HttpClientError {}
89
90/// Posts a JSON payload through the network boundary using the configured timeout.
91pub async fn post_json(
92    config: &HttpConfig,
93    url: &str,
94    payload: &Value,
95) -> Result<Value, HttpClientError> {
96    let request = JsonHttpRequest::parse(url)?;
97    let body = serde_json::to_vec(payload).map_err(HttpClientError::ResponseJson)?;
98    let response = tokio::time::timeout(config.request_timeout, send_json_request(request, body))
99        .await
100        .map_err(|_| HttpClientError::Timeout)??;
101
102    serde_json::from_slice(&response).map_err(HttpClientError::ResponseJson)
103}
104
105/// Posts JSON through the raw worker HTTP helper after outbound QoS admission.
106pub async fn post_json_with_qos(
107    config: &HttpConfig,
108    qos: &QosRuntime,
109    policy: &QosPolicy,
110    url: &str,
111    payload: &Value,
112) -> Result<Value, HttpClientError> {
113    let permit = if qos_request_context_active() {
114        None
115    } else {
116        Some(
117            qos.admit_request(policy)
118                .map_err(HttpClientError::QosRejected)?,
119        )
120    };
121    let result = post_json(config, url, payload).await;
122    drop(permit);
123    if matches!(result, Err(HttpClientError::Timeout)) {
124        qos.record_timed_out();
125    }
126
127    result
128}
129
130struct JsonHttpRequest {
131    host: String,
132    port: u16,
133    path: String,
134}
135
136impl JsonHttpRequest {
137    fn parse(value: &str) -> Result<Self, HttpClientError> {
138        let remainder = value
139            .strip_prefix("http://")
140            .ok_or_else(|| HttpClientError::InvalidUrl(value.to_owned()))?;
141        let (authority, path) = remainder
142            .split_once('/')
143            .map_or((remainder, "/"), |(authority, path)| {
144                (authority, path.trim_start_matches('/'))
145            });
146        if authority.is_empty() {
147            return Err(HttpClientError::InvalidUrl(value.to_owned()));
148        }
149        let (host, port) = authority
150            .rsplit_once(':')
151            .map(|(host, port)| {
152                let parsed_port = port
153                    .parse::<u16>()
154                    .map_err(|_| HttpClientError::InvalidUrl(value.to_owned()))?;
155                Ok((host.to_owned(), parsed_port))
156            })
157            .unwrap_or_else(|| Ok((authority.to_owned(), 80)))?;
158        if host.is_empty() || port == 0 {
159            return Err(HttpClientError::InvalidUrl(value.to_owned()));
160        }
161        let path = if path.is_empty() {
162            "/".to_owned()
163        } else {
164            format!("/{path}")
165        };
166
167        Ok(Self { host, port, path })
168    }
169}
170
171async fn send_json_request(
172    request: JsonHttpRequest,
173    body: Vec<u8>,
174) -> Result<Vec<u8>, HttpClientError> {
175    let mut stream = tokio::net::TcpStream::connect((request.host.as_str(), request.port))
176        .await
177        .map_err(HttpClientError::Io)?;
178    let head = format!(
179        "POST {} HTTP/1.1\r\nHost: {}\r\nContent-Type: application/json\r\nAccept: application/json\r\nConnection: close\r\nContent-Length: {}\r\n\r\n",
180        request.path,
181        request.host,
182        body.len()
183    );
184    stream
185        .write_all(head.as_bytes())
186        .await
187        .map_err(HttpClientError::Io)?;
188    stream.write_all(&body).await.map_err(HttpClientError::Io)?;
189    stream.shutdown().await.map_err(HttpClientError::Io)?;
190    let mut response = Vec::new();
191    stream
192        .read_to_end(&mut response)
193        .await
194        .map_err(HttpClientError::Io)?;
195    parse_http_response(response)
196}
197
198fn parse_http_response(response: Vec<u8>) -> Result<Vec<u8>, HttpClientError> {
199    let Some(header_end) = response.windows(4).position(|window| window == b"\r\n\r\n") else {
200        return Err(HttpClientError::InvalidResponse);
201    };
202    let headers = std::str::from_utf8(&response[..header_end])
203        .map_err(|_| HttpClientError::InvalidResponse)?;
204    let status = headers
205        .lines()
206        .next()
207        .and_then(|line| line.split_whitespace().nth(1))
208        .and_then(|value| value.parse::<u16>().ok())
209        .ok_or(HttpClientError::InvalidResponse)?;
210    if !(200..300).contains(&status) {
211        return Err(HttpClientError::ResponseStatus(status));
212    }
213
214    Ok(response[header_end + 4..].to_vec())
215}
216
217#[cfg(test)]
218#[path = "outbound_tests.rs"]
219mod outbound_tests;
relay_knowledge/net/http/outbound.rs

relay_knowledge/net/http/
outbound.rs
