relay_knowledge/net/http/
mod.rs

1//! HTTP client and server runtime owned by the network boundary.
2//!
3//! This module owns validated HTTP configuration, outbound JSON client policy,
4//! the bounded raw JSON POST helper, async Axum router serving, request body
5//! limits, per-request timeouts, graceful shutdown, and QoS-gated listener
6//! admission. Higher layers should use these APIs instead of constructing
7//! sockets, listeners, HTTP clients, or HTTP server loops directly.
8
9use std::{
10    convert::Infallible,
11    error::Error,
12    fmt,
13    future::{Future, IntoFuture, Ready, ready},
14    io,
15    net::IpAddr,
16    pin::Pin,
17    sync::{
18        Arc,
19        atomic::{AtomicU64, Ordering},
20    },
21    task::{Context, Poll},
22    time::Duration,
23};
24
25mod outbound;
26mod qos_admission;
27mod qos_client;
28
29use axum::{
30    Router,
31    extract::Request,
32    http::StatusCode,
33    response::{IntoResponse, Response},
34    serve::{IncomingStream, Listener},
35};
36use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
37use tower::Service;
38
39use qos_admission::QosRequestLayer;
40
41use crate::{
42    env::NetworkEnvOverrides,
43    net::qos::{QosPermit, QosPolicy, QosRuntime},
44};
45
46pub use outbound::{
47    HttpClientError, OutboundClientError, outbound_json_client, outbound_json_client_with_policy,
48    post_json, post_json_with_qos,
49};
50pub use qos_client::{QosHttpClientError, QosHttpResponse, send_request_with_qos};
51
52tokio::task_local! {
53    static QOS_REQUEST_CONTEXT: ();
54}
55
56pub const DEFAULT_HTTP_BIND: &str = "127.0.0.1:8791";
57pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
58pub const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
59pub const DEFAULT_MAX_BODY_BYTES: u64 = 1_048_576;
60pub const DEFAULT_SSL_VERIFY: bool = true;
61
62/// Event-driven HTTP configuration for inbound and outbound adapters.
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub struct HttpConfig {
65    pub bind_address: HttpBindAddress,
66    pub request_timeout: Duration,
67    pub graceful_shutdown_timeout: Duration,
68    pub max_request_body_bytes: u64,
69    pub proxy: HttpProxyConfig,
70}
71
72/// Validated HTTP bind address in `host:port` form.
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub struct HttpBindAddress {
75    value: String,
76    port: u16,
77}
78
79impl HttpBindAddress {
80    /// Parses a host or IP literal with an explicit non-zero port.
81    pub fn parse(value: &str) -> Result<Self, HttpConfigError> {
82        let trimmed = value.trim();
83        if trimmed.is_empty() {
84            return Err(HttpConfigError::InvalidBindAddress {
85                value: value.to_owned(),
86            });
87        }
88
89        if let Ok(socket_addr) = trimmed.parse::<std::net::SocketAddr>() {
90            return Self::from_parts(trimmed.to_owned(), socket_addr.port());
91        }
92
93        let Some((host, port)) = trimmed.rsplit_once(':') else {
94            return Err(HttpConfigError::InvalidBindAddress {
95                value: value.to_owned(),
96            });
97        };
98
99        if host.is_empty() || host.contains('/') || host.contains(char::is_whitespace) {
100            return Err(HttpConfigError::InvalidBindAddress {
101                value: value.to_owned(),
102            });
103        }
104
105        let port = port
106            .parse::<u16>()
107            .map_err(|_| HttpConfigError::InvalidBindAddress {
108                value: value.to_owned(),
109            })?;
110
111        Self::from_parts(trimmed.to_owned(), port)
112    }
113
114    /// Returns the explicit TCP port.
115    pub const fn port(&self) -> u16 {
116        self.port
117    }
118
119    fn from_parts(value: String, port: u16) -> Result<Self, HttpConfigError> {
120        if port == 0 {
121            return Err(HttpConfigError::EphemeralPort);
122        }
123
124        Ok(Self { value, port })
125    }
126}
127
128impl fmt::Display for HttpBindAddress {
129    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
130        formatter.write_str(&self.value)
131    }
132}
133
134/// Returns whether a listener may accept non-local clients under the access policy.
135pub fn remote_clients_allowed(config: &HttpConfig, allow_remote_clients: bool) -> bool {
136    allow_remote_clients || is_local_bind(&config.bind_address.to_string())
137}
138
139/// Outbound HTTP proxy and TLS verification policy.
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct HttpProxyConfig {
142    pub proxy: Option<String>,
143    pub no_proxy_rules: Vec<String>,
144    pub ssl_verify: bool,
145}
146
147impl HttpProxyConfig {
148    /// Validates proxy URL shape and no-proxy entries without exposing credentials.
149    pub fn new(
150        proxy: Option<String>,
151        no_proxy_rules: Vec<String>,
152        ssl_verify: bool,
153    ) -> Result<Self, HttpConfigError> {
154        if let Some(proxy_url) = proxy.as_deref() {
155            validate_proxy_url(proxy_url)?;
156        }
157
158        for rule in &no_proxy_rules {
159            if rule.trim().is_empty() {
160                return Err(HttpConfigError::EmptyNoProxyRule);
161            }
162        }
163
164        Ok(Self {
165            proxy,
166            no_proxy_rules,
167            ssl_verify,
168        })
169    }
170
171    /// Applies proxy, no-proxy, and TLS verification environment overrides.
172    pub fn from_overrides(overrides: &NetworkEnvOverrides) -> Result<Self, HttpConfigError> {
173        Self::new(
174            overrides.proxy.clone(),
175            parse_no_proxy_rules(overrides.no_proxy.as_deref())?,
176            overrides.ssl_verify.unwrap_or(DEFAULT_SSL_VERIFY),
177        )
178    }
179
180    /// Returns whether outbound HTTP should use a proxy.
181    pub fn is_proxy_configured(&self) -> bool {
182        self.proxy.is_some()
183    }
184}
185
186impl HttpConfig {
187    /// Builds HTTP config while enforcing bounded request and shutdown behavior.
188    pub fn new(
189        bind_address: HttpBindAddress,
190        request_timeout: Duration,
191        graceful_shutdown_timeout: Duration,
192        max_request_body_bytes: u64,
193        proxy: HttpProxyConfig,
194    ) -> Result<Self, HttpConfigError> {
195        if request_timeout.is_zero() {
196            return Err(HttpConfigError::ZeroDuration {
197                field: "request_timeout",
198            });
199        }
200
201        if graceful_shutdown_timeout.is_zero() {
202            return Err(HttpConfigError::ZeroDuration {
203                field: "graceful_shutdown_timeout",
204            });
205        }
206
207        if max_request_body_bytes == 0 {
208            return Err(HttpConfigError::ZeroMaxBodyBytes);
209        }
210
211        Ok(Self {
212            bind_address,
213            request_timeout,
214            graceful_shutdown_timeout,
215            max_request_body_bytes,
216            proxy,
217        })
218    }
219
220    /// Applies environment overrides to the default local HTTP policy.
221    pub fn from_overrides(overrides: &NetworkEnvOverrides) -> Result<Self, HttpConfigError> {
222        let bind_value = overrides.http_bind.as_deref().unwrap_or(DEFAULT_HTTP_BIND);
223        let bind_address = HttpBindAddress::parse(bind_value)?;
224        let request_timeout = overrides
225            .http_request_timeout_ms
226            .map(Duration::from_millis)
227            .unwrap_or(DEFAULT_REQUEST_TIMEOUT);
228        let shutdown_timeout = overrides
229            .http_shutdown_timeout_ms
230            .map(Duration::from_millis)
231            .unwrap_or(DEFAULT_SHUTDOWN_TIMEOUT);
232        let max_body_bytes = overrides
233            .http_max_body_bytes
234            .unwrap_or(DEFAULT_MAX_BODY_BYTES);
235        let proxy = HttpProxyConfig::from_overrides(overrides)?;
236
237        Self::new(
238            bind_address,
239            request_timeout,
240            shutdown_timeout,
241            max_body_bytes,
242            proxy,
243        )
244    }
245}
246
247/// HTTP configuration validation error.
248#[derive(Debug, Clone, PartialEq, Eq)]
249pub enum HttpConfigError {
250    InvalidBindAddress { value: String },
251    EphemeralPort,
252    ZeroDuration { field: &'static str },
253    ZeroMaxBodyBytes,
254    InvalidProxyUrl,
255    EmptyNoProxyRule,
256}
257
258impl fmt::Display for HttpConfigError {
259    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
260        match self {
261            Self::InvalidBindAddress { value } => {
262                write!(formatter, "bind address '{value}' is not host:port")
263            }
264            Self::EphemeralPort => write!(formatter, "bind address must use an explicit port"),
265            Self::ZeroDuration { field } => write!(formatter, "{field} must be greater than zero"),
266            Self::ZeroMaxBodyBytes => write!(
267                formatter,
268                "max request body bytes must be greater than zero"
269            ),
270            Self::InvalidProxyUrl => write!(
271                formatter,
272                "proxy must use http:// or https:// and include a host"
273            ),
274            Self::EmptyNoProxyRule => write!(formatter, "no-proxy entries must not be empty"),
275        }
276    }
277}
278
279impl Error for HttpConfigError {}
280
281/// Error raised while serving an event-driven HTTP adapter.
282#[derive(Debug)]
283pub enum HttpServeError {
284    Bind(std::io::Error),
285    Serve(std::io::Error),
286    ShutdownTimeout,
287}
288
289impl fmt::Display for HttpServeError {
290    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
291        match self {
292            Self::Bind(error) => write!(formatter, "failed to bind HTTP listener: {error}"),
293            Self::Serve(error) => write!(formatter, "HTTP server failed: {error}"),
294            Self::ShutdownTimeout => write!(formatter, "HTTP graceful shutdown timed out"),
295        }
296    }
297}
298
299impl Error for HttpServeError {}
300
301/// Stable identifier assigned to an accepted HTTP connection.
302#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
303pub struct HttpConnectionId(u64);
304
305impl HttpConnectionId {
306    /// Returns the numeric connection identifier for request correlation.
307    pub const fn get(self) -> u64 {
308        self.0
309    }
310}
311
312/// Starts an async HTTP server with graceful shutdown under the network boundary.
313pub async fn serve_router(
314    router: Router,
315    config: HttpConfig,
316    shutdown: impl Future<Output = ()> + Send + 'static,
317) -> Result<(), HttpServeError> {
318    let listener = tokio::net::TcpListener::bind(config.bind_address.to_string())
319        .await
320        .map_err(HttpServeError::Bind)?;
321
322    serve_listener(listener, router, config, None, shutdown).await
323}
324
325/// Starts an async HTTP server whose accepted connections consume QoS permits.
326pub async fn serve_router_with_qos(
327    router: Router,
328    config: HttpConfig,
329    qos: QosRuntime,
330    policy: QosPolicy,
331    shutdown: impl Future<Output = ()> + Send + 'static,
332) -> Result<(), HttpServeError> {
333    let listener = tokio::net::TcpListener::bind(config.bind_address.to_string())
334        .await
335        .map_err(HttpServeError::Bind)?;
336    let listener = QosTcpListener::new(listener, qos.clone(), policy);
337
338    serve_listener(listener, router, config, Some(qos), shutdown).await
339}
340
341/// Adds per-request QoS admission to an Axum router without opening sockets.
342pub fn router_with_qos_request_admission(
343    router: Router,
344    qos: QosRuntime,
345    policy: QosPolicy,
346) -> Router {
347    router.layer(QosRequestLayer::new(qos, policy))
348}
349
350async fn serve_listener<L>(
351    listener: L,
352    router: Router,
353    config: HttpConfig,
354    timeout_qos: Option<QosRuntime>,
355    shutdown: impl Future<Output = ()> + Send + 'static,
356) -> Result<(), HttpServeError>
357where
358    L: Listener,
359    L::Addr: fmt::Debug,
360{
361    let (shutdown_started, mut shutdown_observed) = tokio::sync::watch::channel(false);
362    let graceful_shutdown = async move {
363        shutdown.await;
364        let _ = shutdown_started.send(true);
365    };
366    let server = axum::serve(
367        listener,
368        HttpMakeService::new(router, config.request_timeout, timeout_qos),
369    )
370    .with_graceful_shutdown(graceful_shutdown)
371    .into_future();
372
373    tokio::pin!(server);
374    tokio::select! {
375        result = &mut server => result.map_err(HttpServeError::Serve),
376        changed = shutdown_observed.changed() => {
377            let _ = changed;
378            match tokio::time::timeout(config.graceful_shutdown_timeout, &mut server).await {
379                Ok(result) => result.map_err(HttpServeError::Serve),
380                Err(_) => Err(HttpServeError::ShutdownTimeout),
381            }
382        }
383    }
384}
385
386struct HttpMakeService {
387    router: Router,
388    request_timeout: Duration,
389    next_connection_id: Arc<AtomicU64>,
390    timeout_qos: Option<QosRuntime>,
391}
392
393impl HttpMakeService {
394    fn new(router: Router, request_timeout: Duration, timeout_qos: Option<QosRuntime>) -> Self {
395        Self {
396            router,
397            request_timeout,
398            next_connection_id: Arc::new(AtomicU64::new(1)),
399            timeout_qos,
400        }
401    }
402}
403
404impl<'a, L> Service<IncomingStream<'a, L>> for HttpMakeService
405where
406    L: Listener,
407{
408    type Response = HttpConnectionService<Router>;
409    type Error = Infallible;
410    type Future = Ready<Result<Self::Response, Self::Error>>;
411
412    fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
413        Poll::Ready(Ok(()))
414    }
415
416    fn call(&mut self, _target: IncomingStream<'a, L>) -> Self::Future {
417        let connection_id =
418            HttpConnectionId(self.next_connection_id.fetch_add(1, Ordering::Relaxed));
419        ready(Ok(HttpConnectionService::new(
420            self.router.clone(),
421            connection_id,
422            self.request_timeout,
423            self.timeout_qos.clone(),
424        )))
425    }
426}
427
428struct HttpConnectionService<S> {
429    inner: S,
430    connection_id: HttpConnectionId,
431    request_timeout: Duration,
432    timeout_qos: Option<QosRuntime>,
433}
434
435impl<S> HttpConnectionService<S> {
436    fn new(
437        inner: S,
438        connection_id: HttpConnectionId,
439        request_timeout: Duration,
440        timeout_qos: Option<QosRuntime>,
441    ) -> Self {
442        Self {
443            inner,
444            connection_id,
445            request_timeout,
446            timeout_qos,
447        }
448    }
449}
450
451impl<S> Clone for HttpConnectionService<S>
452where
453    S: Clone,
454{
455    fn clone(&self) -> Self {
456        Self {
457            inner: self.inner.clone(),
458            connection_id: self.connection_id,
459            request_timeout: self.request_timeout,
460            timeout_qos: self.timeout_qos.clone(),
461        }
462    }
463}
464
465impl<S> Service<Request> for HttpConnectionService<S>
466where
467    S: Service<Request, Response = Response, Error = Infallible> + Send,
468    S::Future: Send + 'static,
469{
470    type Response = Response;
471    type Error = Infallible;
472    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
473
474    fn poll_ready(&mut self, context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
475        self.inner.poll_ready(context)
476    }
477
478    fn call(&mut self, mut request: Request) -> Self::Future {
479        request.extensions_mut().insert(self.connection_id);
480        let future = self.inner.call(request);
481        let request_timeout = self.request_timeout;
482        let timeout_qos = self.timeout_qos.clone();
483        Box::pin(async move {
484            match tokio::time::timeout(request_timeout, future).await {
485                Ok(result) => result,
486                Err(_) => {
487                    if let Some(qos) = timeout_qos {
488                        qos.record_timed_out();
489                    }
490                    Ok((StatusCode::REQUEST_TIMEOUT, "request timed out").into_response())
491                }
492            }
493        })
494    }
495}
496
497pub(crate) fn qos_request_context_active() -> bool {
498    QOS_REQUEST_CONTEXT.try_with(|_| ()).is_ok()
499}
500
501struct QosTcpListener {
502    inner: tokio::net::TcpListener,
503    qos: QosRuntime,
504    policy: QosPolicy,
505}
506
507impl QosTcpListener {
508    fn new(inner: tokio::net::TcpListener, qos: QosRuntime, policy: QosPolicy) -> Self {
509        Self { inner, qos, policy }
510    }
511}
512
513impl Listener for QosTcpListener {
514    type Io = QosTcpStream;
515    type Addr = std::net::SocketAddr;
516
517    async fn accept(&mut self) -> (Self::Io, Self::Addr) {
518        loop {
519            match self.inner.accept().await {
520                Ok((stream, address)) => match self.qos.admit_connection(&self.policy) {
521                    Ok(permit) => {
522                        return (
523                            QosTcpStream {
524                                inner: stream,
525                                _permit: permit,
526                            },
527                            address,
528                        );
529                    }
530                    Err(_) => {
531                        self.qos.record_dropped();
532                        drop(stream)
533                    }
534                },
535                Err(_) => tokio::time::sleep(Duration::from_secs(1)).await,
536            }
537        }
538    }
539
540    fn local_addr(&self) -> io::Result<Self::Addr> {
541        self.inner.local_addr()
542    }
543}
544
545struct QosTcpStream {
546    inner: tokio::net::TcpStream,
547    _permit: QosPermit,
548}
549
550impl AsyncRead for QosTcpStream {
551    fn poll_read(
552        mut self: Pin<&mut Self>,
553        context: &mut Context<'_>,
554        buffer: &mut ReadBuf<'_>,
555    ) -> Poll<io::Result<()>> {
556        Pin::new(&mut self.inner).poll_read(context, buffer)
557    }
558}
559
560impl AsyncWrite for QosTcpStream {
561    fn poll_write(
562        mut self: Pin<&mut Self>,
563        context: &mut Context<'_>,
564        buffer: &[u8],
565    ) -> Poll<io::Result<usize>> {
566        Pin::new(&mut self.inner).poll_write(context, buffer)
567    }
568
569    fn poll_flush(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<io::Result<()>> {
570        Pin::new(&mut self.inner).poll_flush(context)
571    }
572
573    fn poll_shutdown(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<io::Result<()>> {
574        Pin::new(&mut self.inner).poll_shutdown(context)
575    }
576}
577
578fn validate_proxy_url(value: &str) -> Result<(), HttpConfigError> {
579    let Some((scheme, remainder)) = value.split_once("://") else {
580        return Err(HttpConfigError::InvalidProxyUrl);
581    };
582
583    if !matches!(scheme, "http" | "https") {
584        return Err(HttpConfigError::InvalidProxyUrl);
585    }
586
587    let authority = remainder.split('/').next().unwrap_or_default();
588    if authority.is_empty() || authority.starts_with('@') {
589        return Err(HttpConfigError::InvalidProxyUrl);
590    }
591    let host_port = authority
592        .rsplit_once('@')
593        .map_or(authority, |(_, host)| host);
594    let host = if let Some(remainder) = host_port.strip_prefix('[') {
595        remainder.split_once(']').map_or("", |(host, _)| host)
596    } else {
597        host_port.split(':').next().unwrap_or_default()
598    };
599    if host.is_empty() {
600        return Err(HttpConfigError::InvalidProxyUrl);
601    }
602
603    Ok(())
604}
605
606fn parse_no_proxy_rules(value: Option<&str>) -> Result<Vec<String>, HttpConfigError> {
607    value
608        .map(|rules| {
609            rules
610                .split(',')
611                .map(str::trim)
612                .map(|rule| {
613                    if rule.is_empty() {
614                        Err(HttpConfigError::EmptyNoProxyRule)
615                    } else {
616                        Ok(rule.to_owned())
617                    }
618                })
619                .collect()
620        })
621        .unwrap_or_else(|| Ok(Vec::new()))
622}
623
624fn is_local_bind(bind: &str) -> bool {
625    is_loopback_host(authority_host(bind))
626}
627
628fn authority_host(authority: &str) -> &str {
629    if let Some(remainder) = authority.strip_prefix('[') {
630        return remainder
631            .find(']')
632            .map_or(authority, |index| &remainder[..index]);
633    }
634
635    authority
636        .rsplit_once(':')
637        .map_or(authority, |(host, _)| host)
638}
639
640fn is_loopback_host(host: &str) -> bool {
641    host.eq_ignore_ascii_case("localhost")
642        || host
643            .parse::<IpAddr>()
644            .is_ok_and(|address| address.is_loopback())
645}
646
647#[cfg(test)]
648#[path = "mod_tests.rs"]
649mod tests;
relay_knowledge/net/http/mod.rs

relay_knowledge/net/http/
mod.rs
