iroh_http_core/http/client.rs
1//! Outgoing HTTP request — pure-Rust `fetch_request()` implementation.
2//!
3//! HTTP/1.1 framing is delegated entirely to hyper. Iroh's QUIC stream pair
4//! is wrapped in `IrohStream` and handed to hyper's client connection API.
5//!
6//! Slice D (#186) split the original FFI-shaped `fetch(endpoint, &str, ...)`
7//! into two layers:
8//!
9//! - [`fetch_request`] — pure-Rust API: takes a fully-formed
10//! [`hyper::Request<Body>`] and an [`iroh::EndpointAddr`], returns
11//! [`hyper::Response<Body>`] with a typed [`FetchError`]. No `u64`
12//! handles, no `BodyReader`, no string parsing of error messages, no
13//! imports from `crate::ffi`.
14//! - [`crate::ffi::fetch::fetch`] — FFI-shaped wrapper that builds the
15//! `Request<Body>` from flat strings, calls [`fetch_request`], and
16//! translates the response into a [`crate::FfiResponse`].
17
18use hyper_util::rt::TokioIo;
19
20use crate::{
21 http::{server::stack::StackConfig, transport::io::IrohStream},
22 Body, IrohEndpoint, ALPN,
23};
24
25// ── Typed fetch error ────────────────────────────────────────────────────────
26
27/// Typed error returned by the pure-Rust [`fetch_request`] API.
28///
29/// Header-oversize detection lives **above** this layer in
30/// [`crate::ffi::fetch`], which compares the assembled response head
31/// byte-count against the endpoint's `max_header_size`. That check is
32/// deterministic and does not depend on hyper's error wording, so this
33/// enum no longer disambiguates header parse failures from other
34/// transport errors — they all surface as `ConnectionFailed`. Failures yielded
35/// by the caller's request body are identified separately as
36/// [`FetchError::RequestBodyFailed`].
37#[derive(Debug)]
38#[non_exhaustive]
39pub enum FetchError {
40 /// Connection setup, hyper handshake, or send-request transport
41 /// failure. Wraps the underlying [`hyper::Error`] when one is
42 /// available so callers can downcast instead of substring-matching.
43 ConnectionFailed {
44 detail: String,
45 source: Option<hyper::Error>,
46 },
47 /// The caller-supplied request body yielded an error while Hyper was
48 /// streaming it. This is a local request-production failure, not evidence
49 /// that the pooled transport is unhealthy.
50 RequestBodyFailed {
51 detail: String,
52 source: hyper::Error,
53 },
54 /// Response head exceeded the endpoint's `max_header_size` budget.
55 /// Produced only by [`crate::ffi::fetch`]'s post-receive byte-count
56 /// check; [`fetch_request`] itself never emits this variant.
57 HeaderTooLarge { detail: String },
58 /// Response body exceeded the configured byte limit. Surfaced by the
59 /// FFI wrapper after the body is drained.
60 BodyTooLarge,
61 /// `cfg.timeout` elapsed before the response head arrived.
62 Timeout,
63 /// Caller dropped the future or signalled cancellation via the FFI
64 /// fetch token.
65 Cancelled,
66 /// Bug or unexpected internal failure (request build, body wrap, …).
67 Internal(String),
68}
69
70impl std::fmt::Display for FetchError {
71 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72 match self {
73 FetchError::ConnectionFailed { detail, .. } => {
74 write!(f, "connection failed: {detail}")
75 }
76 FetchError::RequestBodyFailed { detail, .. } => {
77 write!(f, "request body failed: {detail}")
78 }
79 FetchError::HeaderTooLarge { detail } => {
80 write!(f, "response header too large: {detail}")
81 }
82 FetchError::BodyTooLarge => f.write_str("response body too large"),
83 FetchError::Timeout => f.write_str("request timed out"),
84 FetchError::Cancelled => f.write_str("request cancelled"),
85 FetchError::Internal(msg) => write!(f, "internal error: {msg}"),
86 }
87 }
88}
89
90impl std::error::Error for FetchError {
91 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
92 match self {
93 FetchError::ConnectionFailed {
94 source: Some(s), ..
95 } => Some(s),
96 FetchError::RequestBodyFailed { source, .. } => Some(source),
97 _ => None,
98 }
99 }
100}
101
102/// Marker attached only to errors yielded by the caller-supplied request body.
103///
104/// Hyper's broad `is_user()` category also includes dispatch-channel failures
105/// caused by a dead connection task, so it cannot distinguish body production
106/// from transport death. This private marker survives in Hyper's source chain
107/// and gives retry/eviction logic an exact origin instead.
108#[derive(Debug)]
109struct CallerRequestBodyError {
110 source: crate::BoxError,
111}
112
113impl std::fmt::Display for CallerRequestBodyError {
114 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
115 write!(f, "caller request body failed: {}", self.source)
116 }
117}
118
119impl std::error::Error for CallerRequestBodyError {
120 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
121 Some(self.source.as_ref())
122 }
123}
124
125/// Tag caller-body failures and preserve non-DATA frames from an exact-zero
126/// DATA body that has not reached end-of-stream.
127///
128/// Hyper treats an exact-zero size hint as a framing decision and may not poll
129/// the body at all. That is correct for a body already at end-of-stream, but a
130/// still-live body can legally yield trailers or an error frame. Masking only
131/// that hint as unknown makes Hyper poll those frames without buffering or
132/// otherwise changing the request body.
133struct MarkCallerRequestBody {
134 body: Body,
135 mask_zero_data_hint: bool,
136}
137
138impl http_body::Body for MarkCallerRequestBody {
139 type Data = bytes::Bytes;
140 type Error = CallerRequestBodyError;
141
142 fn poll_frame(
143 self: std::pin::Pin<&mut Self>,
144 cx: &mut std::task::Context<'_>,
145 ) -> std::task::Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
146 let this = self.get_mut();
147 match std::pin::Pin::new(&mut this.body).poll_frame(cx) {
148 std::task::Poll::Ready(Some(Err(source))) => {
149 std::task::Poll::Ready(Some(Err(CallerRequestBodyError { source })))
150 }
151 std::task::Poll::Ready(Some(Ok(frame))) => std::task::Poll::Ready(Some(Ok(frame))),
152 std::task::Poll::Ready(None) => std::task::Poll::Ready(None),
153 std::task::Poll::Pending => std::task::Poll::Pending,
154 }
155 }
156
157 fn is_end_stream(&self) -> bool {
158 self.body.is_end_stream()
159 }
160
161 fn size_hint(&self) -> http_body::SizeHint {
162 if self.mask_zero_data_hint {
163 http_body::SizeHint::default()
164 } else {
165 self.body.size_hint()
166 }
167 }
168}
169
170/// Return the original caller-body failure detail only when the private origin
171/// marker is present in Hyper's error source chain.
172fn caller_request_body_error_detail(error: &hyper::Error) -> Option<String> {
173 let mut current: Option<&(dyn std::error::Error + 'static)> = Some(error);
174 while let Some(source) = current {
175 if let Some(body_error) = source.downcast_ref::<CallerRequestBodyError>() {
176 return Some(body_error.source.to_string());
177 }
178 current = source.source();
179 }
180 None
181}
182
183// ── Pure-Rust fetch API ──────────────────────────────────────────────────────
184
185/// Pure-Rust outbound entry — the canonical client API.
186///
187/// Establishes (or reuses, via [`crate::http::transport::pool`]) an Iroh
188/// QUIC connection to `addr`, runs hyper's HTTP/1.1 client handshake on
189/// a freshly opened bidirectional stream, dispatches `req` through the
190/// shared client tower stack ([`crate::http::server::stack::build_client_stack`]),
191/// and returns the response.
192///
193/// The returned [`hyper::Response<Body>`] streams its body lazily; the
194/// caller is responsible for draining it.
195///
196/// # Errors
197///
198/// Returns [`FetchError::Timeout`] if `cfg.timeout` is set and elapsed
199/// before the response head arrived. Connection / handshake / transport
200/// failures map to [`FetchError::ConnectionFailed`]. A caller-supplied request
201/// body error maps to [`FetchError::RequestBodyFailed`] and retains the Hyper
202/// and original body-error source chain.
203pub async fn fetch_request(
204 endpoint: &IrohEndpoint,
205 addr: &iroh::EndpointAddr,
206 req: hyper::Request<Body>,
207 cfg: &StackConfig,
208) -> Result<hyper::Response<Body>, FetchError> {
209 let node_id = addr.id;
210
211 // P2/L1: the dial_seq of the connection the most recent attempt actually
212 // used. Eviction is scoped to this so we only ever close the connection
213 // that failed — never a fresh replacement a concurrent request dialed under
214 // the same key. A monotonic dial_seq (not a reusable stable_id pointer)
215 // makes that guarantee ABA-proof. `None` = no connection was obtained
216 // (connect failed / timed out before dialing), in which case eviction is a
217 // no-op.
218 let last_dial_seq: std::sync::Mutex<Option<u64>> = std::sync::Mutex::new(None);
219
220 // Split the request so a transparent retry can rebuild it. A request whose
221 // body is empty (exact size 0 — every GET/HEAD/DELETE the FFI layer builds
222 // with `Body::empty()`) can be reconstructed byte-for-byte on a fresh
223 // connection. A streaming body may have been partially read by a failed
224 // attempt, so it is never resent.
225 //
226 // F18: attempt 1 retains the original request parts and real body. The body
227 // gets only a transparent error-origin marker (and, for a still-live
228 // exact-zero DATA body, an unknown size hint so Hyper polls trailers/errors).
229 // Only the transparent retry, gated on `retry_safe`, is rebuilt from the
230 // cloned method/uri/version/headers with a fresh `Body::empty()`.
231 let (parts, body) = req.into_parts();
232 let body_exact_size = http_body::Body::size_hint(&body).exact();
233 let body_at_end = http_body::Body::is_end_stream(&body);
234 // `SizeHint::exact(0)` constrains DATA bytes only: a body can still yield
235 // trailers or an error frame. `is_end_stream()` is the public guarantee
236 // that no frames remain, which is what makes replacing it with a fresh
237 // `Body::empty()` semantically safe on retry.
238 let body_resendable = body_at_end && body_exact_size == Some(0);
239 let body = if body_at_end {
240 body
241 } else {
242 Body::new(MarkCallerRequestBody {
243 body,
244 mask_zero_data_hint: body_exact_size == Some(0),
245 })
246 };
247 let method = parts.method.clone();
248 // P1: whether it is SAFE to transparently replay this request. RFC 9110
249 // §9.2.2 only permits automatic replay when the method is idempotent AND no
250 // response was received — body-emptiness is NOT a substitute for
251 // idempotency. The FFI builds `Body::empty()` for *any* method (including
252 // POST/PATCH) when no body reader is supplied, and `attempt_send` can
253 // surface `ConnectionFailed` even after the peer's (old) serve loop already
254 // processed the request bytes (the #336 replace-close race). Retrying an
255 // empty-body POST/PATCH would silently double-execute it. Gate the retry on
256 // `Method::is_idempotent()` (GET/HEAD/PUT/DELETE/OPTIONS/TRACE = true;
257 // POST/PATCH = false). The #119 recovery path and all iOS interop traffic
258 // are GET, so this does not regress it.
259 let retry_safe = body_resendable && method.is_idempotent();
260 let uri = parts.uri.clone();
261 let version = parts.version;
262 let headers = parts.headers.clone();
263 // Only used to rebuild a transparent retry — never for attempt 1.
264 let build_retry = |body: Body| -> hyper::Request<Body> {
265 let mut builder = hyper::Request::builder()
266 .method(method.clone())
267 .uri(uri.clone())
268 .version(version);
269 if let Some(h) = builder.headers_mut() {
270 *h = headers.clone();
271 }
272 // Infallible: method/uri/version/headers all came from a valid request.
273 builder
274 .body(body)
275 .expect("rebuild request from valid parts")
276 };
277 // Reassemble the untouched first request without cloning the body.
278 let first_req = hyper::Request::from_parts(parts, body);
279
280 // Whole-fetch timeout budget: bounds attempt + retry together so a single
281 // fetch never exceeds `cfg.timeout` even when it redials once (#336).
282 let retry_seq = async {
283 // ── Attempt 1: on the pooled (possibly reused) connection. ───────────
284 let (result, reused) =
285 attempt_send(endpoint, addr, node_id, first_req, cfg, &last_dial_seq).await;
286
287 let first_err = match result {
288 Ok(resp) => return Ok(resp),
289 Err(e) => e,
290 };
291
292 // Only connection-level failures are retryable: a stale pooled
293 // connection the peer closed (serve-loop replace, idle close) surfaces
294 // as `ConnectionFailed`, and no response head was received so resending
295 // is safe. A `Timeout` means the request was likely accepted but slow —
296 // resending would double the work, so it is not retried here.
297 // Caller-body failures are classified separately via an exact marker
298 // in Hyper's source chain. Do not use `hyper::Error::is_user()` here:
299 // that broad category includes dispatch-channel loss when the
300 // connection task dies, which is precisely a stale transport that
301 // should be evicted and retried when the request is replay-safe.
302 let retryable = matches!(&first_err, FetchError::ConnectionFailed { .. });
303
304 // Evict the (now known-bad) pooled connection so attempt 2 — and any
305 // concurrent/future request — dials a fresh one instead of reusing it.
306 // Scoped to the connection attempt 1 used (P2).
307 if retryable {
308 let failed_seq = *last_dial_seq.lock().unwrap_or_else(|e| e.into_inner());
309 endpoint
310 .pool()
311 .evict_and_close(node_id, ALPN, b"iroh-http fetch failed", failed_seq)
312 .await;
313 }
314
315 // Transparent single retry: only for a *reused* connection (a fresh
316 // dial that already failed would just fail again) carrying a
317 // resendable, idempotent request (`retry_safe`). This makes pooled-
318 // connection reuse across a serve replace transparent (#119) while
319 // preserving the #336 invariant that the peer lands on the new serve
320 // loop via a fresh connection — and never double-executes a
321 // non-idempotent request (P1).
322 if retryable && reused && retry_safe {
323 let (retry_result, _) = attempt_send(
324 endpoint,
325 addr,
326 node_id,
327 build_retry(Body::empty()),
328 cfg,
329 &last_dial_seq,
330 )
331 .await;
332 if let Err(FetchError::ConnectionFailed { .. }) = &retry_result {
333 let failed_seq = *last_dial_seq.lock().unwrap_or_else(|e| e.into_inner());
334 endpoint
335 .pool()
336 .evict_and_close(node_id, ALPN, b"iroh-http fetch retry failed", failed_seq)
337 .await;
338 }
339 return retry_result;
340 }
341
342 Err(first_err)
343 };
344
345 match cfg.timeout {
346 Some(t) => match tokio::time::timeout(t, retry_seq).await {
347 Ok(r) => r,
348 Err(_) => {
349 // Timed out across the whole sequence. F6: invalidate the pool
350 // entry so the next request redials, but do NOT close the shared
351 // QUIC connection — it may still be multiplexing other in-flight
352 // requests on independent bi-streams, and closing it would reset
353 // them (cascading this local timeout into unrelated failures).
354 // This request's own stream is reset when the cancelled
355 // `retry_seq` future drops its send/recv handles. Scoped to the
356 // connection the in-flight attempt used (P2).
357 let failed_seq = *last_dial_seq.lock().unwrap_or_else(|e| e.into_inner());
358 endpoint.pool().evict_only(node_id, ALPN, failed_seq).await;
359 Err(FetchError::Timeout)
360 }
361 },
362 None => retry_seq.await,
363 }
364}
365
366/// One connect→open_bi→handshake→send attempt. Returns the result together
367/// with whether the connection was **reused** from the pool (see
368/// [`crate::http::transport::pool::ConnectionPool::get_or_connect`]), which the
369/// caller uses to decide whether a connection failure is worth a single retry.
370///
371/// Records the [`PooledConnection::dial_seq`](crate::http::transport::pool::PooledConnection::dial_seq)
372/// of the connection actually used into `last_dial_seq` (or `None` if no
373/// connection was obtained) so the caller can scope eviction to the exact
374/// connection that failed (P2/L1).
375async fn attempt_send(
376 endpoint: &IrohEndpoint,
377 addr: &iroh::EndpointAddr,
378 node_id: iroh::PublicKey,
379 req: hyper::Request<Body>,
380 cfg: &StackConfig,
381 last_dial_seq: &std::sync::Mutex<Option<u64>>,
382) -> (Result<hyper::Response<Body>, FetchError>, bool) {
383 let ep_raw = endpoint.raw().clone();
384 let addr_clone = addr.clone();
385 let max_header_size = endpoint.max_header_size();
386
387 let (pooled, reused) = match endpoint
388 .pool()
389 .get_or_connect(node_id, ALPN, || async move {
390 ep_raw
391 .connect(addr_clone, ALPN)
392 .await
393 .map_err(|e| format!("connect: {e}"))
394 })
395 .await
396 {
397 Ok(v) => v,
398 Err(e) => {
399 // No connection was obtained — clear the identity so a subsequent
400 // eviction (e.g. on timeout) is a no-op rather than closing an
401 // unrelated cached connection.
402 *last_dial_seq.lock().unwrap_or_else(|e| e.into_inner()) = None;
403 return (
404 Err(FetchError::ConnectionFailed {
405 detail: e,
406 source: None,
407 }),
408 // A failed connect means nothing was reused.
409 false,
410 );
411 }
412 };
413
414 // Record the connection this attempt is about to use so eviction is scoped
415 // to it and never closes a fresh replacement dialed by a concurrent request.
416 *last_dial_seq.lock().unwrap_or_else(|e| e.into_inner()) = Some(pooled.dial_seq);
417 let conn = pooled.conn.clone();
418
419 let (send, recv) = match conn.open_bi().await {
420 Ok(pair) => pair,
421 Err(e) => {
422 return (
423 Err(FetchError::ConnectionFailed {
424 detail: format!("open_bi: {e}"),
425 source: None,
426 }),
427 reused,
428 )
429 }
430 };
431 let io = TokioIo::new(IrohStream::new(send, recv));
432
433 let (sender, conn_task) = match hyper::client::conn::http1::Builder::new()
434 // hyper requires max_buf_size >= 8192; clamp upward so small
435 // max_header_size values don't panic. Header-size enforcement
436 // happens at the byte-count check in `ffi::fetch` after the
437 // response is returned (deterministic, framing-independent).
438 .max_buf_size(max_header_size.max(8192))
439 .max_headers(128)
440 .handshake::<_, Body>(io)
441 .await
442 {
443 Ok(v) => v,
444 Err(e) => {
445 return (
446 Err(FetchError::ConnectionFailed {
447 detail: format!("hyper handshake: {e}"),
448 source: Some(e),
449 }),
450 reused,
451 )
452 }
453 };
454
455 // Drive the connection state machine in the background.
456 tokio::spawn(conn_task);
457
458 // Dispatch through the shared client stack (Slice B / #184).
459 use tower::ServiceExt;
460 let svc = crate::http::server::stack::build_client_stack(sender, cfg);
461 let result = svc.oneshot(req).await.map_err(|error| {
462 if let Some(detail) = caller_request_body_error_detail(&error) {
463 FetchError::RequestBodyFailed {
464 detail,
465 source: error,
466 }
467 } else {
468 FetchError::ConnectionFailed {
469 detail: format!("send_request: {error}"),
470 source: Some(error),
471 }
472 }
473 });
474 (result, reused)
475}
476
477// `extract_path` and the hyper-body→channel pumps moved to `ffi/fetch.rs`
478// and `ffi/pumps.rs` respectively (Slice D, #186). They were FFI plumbing
479// in shape and use; keeping them in `mod http` was the last reason this
480// module had to import from `crate::ffi`.