iroh_http_core/http/server/
mod.rs

1//! Incoming HTTP request — pure-Rust `serve()` implementation.
2//!
3//! Each accepted QUIC bidirectional stream is driven by hyper's HTTP/1.1
4//! server connection. The user supplies a `tower::Service<Request<Body>,
5//! Response = Response<Body>, Error = Infallible>`; the per-connection
6//! `AddExtensionLayer` makes the authenticated peer id available as a
7//! [`RemoteNodeId`] request extension (closes #177).
8//!
9//! The accept loop body lives in [`accept`]; this file is the public
10//! surface and option-resolution glue, kept close to the axum reference
11//! shape. Sub-modules: [`options`] (`ServeOptions` + defaults), [`handle`]
12//! (`ServeHandle`), [`error_layer`] (tower → HTTP error converter),
13//! [`pipeline`] / [`stack`] / [`lifecycle`] (per-bistream chain and RAII
14//! guards).
15//!
16//! The FFI-shaped callback API ([`crate::ffi::dispatcher::ffi_serve_with_callback`])
17//! is one specific consumer of this entry — it constructs an
18//! `IrohHttpService` around the JS callback and hands it in like any
19//! other service.
20
21pub(crate) mod accept;
22pub(crate) mod error_layer;
23pub(crate) mod handle;
24pub(crate) mod lifecycle;
25pub(crate) mod options;
26pub(crate) mod pipeline;
27pub(crate) mod stack;
28
29use std::{sync::Arc, time::Duration};
30
31use tower::Service;
32
33use crate::{Body, ConnectionEvent, IrohEndpoint};
34
35use self::accept::{accept_loop, AcceptConfig};
36use self::options::{
37    DEFAULT_CONCURRENCY, DEFAULT_DRAIN_TIMEOUT_MS, DEFAULT_MAX_CONNECTIONS_PER_PEER,
38    DEFAULT_MAX_REQUEST_BODY_BYTES, DEFAULT_REQUEST_TIMEOUT_MS,
39};
40
41// Re-exported from sub-modules so external paths
42// (`crate::http::server::ServeOptions`, `…::ServeHandle`,
43// `…::HandleLayerErrorLayer`, `…::DEFAULT_MAX_RESPONSE_BODY_BYTES`) stay
44// unchanged after Slice C.7 split.
45pub(crate) use self::error_layer::HandleLayerErrorLayer;
46pub use self::handle::ServeHandle;
47pub use self::options::ServeOptions;
48pub(crate) use self::options::DEFAULT_MAX_RESPONSE_BODY_BYTES;
49
50// ── Connection-event callback type ───────────────────────────────────────────
51
52pub(crate) type ConnectionEventFn = Arc<dyn Fn(ConnectionEvent) + Send + Sync>;
53
54/// Authenticated peer node id of the QUIC connection a request arrived
55/// on. Inserted as a request extension by the per-connection
56/// [`tower_http::add_extension::AddExtensionLayer`] in
57/// [`serve_with_events`].
58///
59/// User-facing pure-Rust services consume it with
60/// `req.extensions().get::<RemoteNodeId>()`. Closes #177.
61#[derive(Clone, Debug)]
62pub struct RemoteNodeId(pub Arc<String>);
63
64/// Pure-Rust serve entry — convenience 3-arg wrapper that omits the
65/// connection-event callback. Equivalent to `serve_with_events(ep,
66/// opts, svc, None)`. The endpoint owns the returned serve cycle immediately,
67/// so [`IrohEndpoint::stop_serve`] works without a separate handle-registration
68/// call.
69pub fn serve<S>(endpoint: IrohEndpoint, options: ServeOptions, svc: S) -> ServeHandle
70where
71    S: Service<
72            hyper::Request<Body>,
73            Response = hyper::Response<Body>,
74            Error = std::convert::Infallible,
75        > + Clone
76        + Send
77        + Sync
78        + 'static,
79    S::Future: Send + 'static,
80{
81    serve_with_events(endpoint, options, svc, None)
82}
83
84/// Pure-Rust serve entry — the canonical inbound API.
85///
86/// Accepts any `tower::Service<Request<Body>, Response = Response<Body>,
87/// Error = Infallible>` (`Clone + Send + Sync + 'static`, with `Send`
88/// futures). Each accepted QUIC bidirectional stream is driven by
89/// hyper's HTTP/1.1 server connection through the per-connection tower
90/// stack composed in [`stack::build_stack`]; the user service sees
91/// requests with the authenticated peer id available as a typed
92/// [`RemoteNodeId`] request extension.
93///
94/// `on_connection_event` is called on 0→1 (first connection from a peer)
95/// and 1→0 (last connection from a peer closed) count transitions.
96/// The cycle is installed in `endpoint` before this function returns; adapter
97/// code may retain the returned handle, but need not register it for endpoint
98/// lifecycle methods to reach the server.
99///
100/// # Security
101///
102/// Calling this opens a **public endpoint** on the Iroh overlay network.
103/// Any peer that knows or discovers your node's public key can connect
104/// and send requests. Iroh QUIC authenticates the peer's *identity*
105/// cryptographically, but does not enforce *authorization*. Inspect
106/// [`RemoteNodeId`] in your service and reject untrusted peers.
107pub fn serve_with_events<S>(
108    endpoint: IrohEndpoint,
109    options: ServeOptions,
110    svc: S,
111    on_connection_event: Option<ConnectionEventFn>,
112) -> ServeHandle
113where
114    S: Service<
115            hyper::Request<Body>,
116            Response = hyper::Response<Body>,
117            Error = std::convert::Infallible,
118        > + Clone
119        + Send
120        + Sync
121        + 'static,
122    S::Future: Send + 'static,
123{
124    let cfg = AcceptConfig {
125        max: options.max_concurrency.unwrap_or(DEFAULT_CONCURRENCY),
126        request_timeout: options
127            .request_timeout_ms
128            .map(Duration::from_millis)
129            .unwrap_or(Duration::from_millis(DEFAULT_REQUEST_TIMEOUT_MS)),
130        max_conns_per_peer: options
131            .max_connections_per_peer
132            .unwrap_or(DEFAULT_MAX_CONNECTIONS_PER_PEER),
133        max_request_body_wire_bytes: options
134            .max_request_body_wire_bytes
135            .or(Some(DEFAULT_MAX_REQUEST_BODY_BYTES)),
136        max_request_body_decoded_bytes: options
137            .max_request_body_decoded_bytes
138            .or(Some(DEFAULT_MAX_REQUEST_BODY_BYTES)),
139        max_total_connections: options.max_total_connections,
140        drain_timeout: Duration::from_millis(
141            options.drain_timeout_ms.unwrap_or(DEFAULT_DRAIN_TIMEOUT_MS),
142        ),
143        // Load-shed is opt-out — default `true`.
144        load_shed_enabled: options.load_shed.unwrap_or(true),
145        max_header_size: endpoint.max_header_size(),
146        stack_compression: endpoint.compression().cloned(),
147        // Decompression is opt-out — default `true`.
148        decompression: options.decompression.unwrap_or(true),
149    };
150
151    let shutdown_notify = Arc::new(tokio::sync::Notify::new());
152    let close_flag = Arc::new(std::sync::atomic::AtomicBool::new(false));
153    let close_connections = Arc::new(tokio::sync::Notify::new());
154    let drain_dur = cfg.drain_timeout;
155    let (done_tx, done_rx) = tokio::sync::watch::channel(false);
156
157    // Register the cycle before spawning it. A stop racing after this point
158    // targets this exact token; Notify retains the shutdown permit if the task
159    // has not begun polling yet. The returned handle is a clone of the one in
160    // the endpoint slot, so adapter-side `set_serve_handle` is idempotent.
161    let handle = ServeHandle::pending(
162        endpoint.next_serve_token(),
163        shutdown_notify.clone(),
164        close_flag.clone(),
165        close_connections.clone(),
166        drain_dur,
167        done_rx,
168    );
169    endpoint.register_serve_handle(handle.clone());
170
171    let join = tokio::spawn(accept_loop(
172        endpoint,
173        cfg,
174        svc,
175        on_connection_event,
176        shutdown_notify.clone(),
177        close_flag.clone(),
178        close_connections.clone(),
179        done_tx,
180    ));
181    handle.attach_join(join);
182    handle
183}
iroh_http_core/http/server/mod.rs

iroh_http_core/http/server/
mod.rs
