From 7ff808b03c4859029022f4230c027ddf2a44bc56 Mon Sep 17 00:00:00 2001 From: David Alsh Date: Mon, 19 Oct 2026 08:35:10 +0000 Subject: [PATCH] fix(proxy): stream proxied response bodies instead of buffering Use an io::Error body type so the upstream body can be passed through instead of being collected into memory. --- src/cmd/serve/handler.rs | 3 +-- src/platform/http1/bytes.rs | 8 +++++--- src/platform/http1/http1_server.rs | 10 +++++----- src/platform/http1/res_ext.rs | 24 ++++++++++++++---------- src/platform/proxy/mod.rs | 15 +++------------ 5 files changed, 28 insertions(+), 32 deletions(-) diff --git a/src/cmd/serve/handler.rs b/src/cmd/serve/handler.rs index f2ac00b..7280ea6 100644 --- a/src/cmd/serve/handler.rs +++ b/src/cmd/serve/handler.rs @@ -1,4 +1,3 @@ -use std::convert::Infallible; #[cfg(unix)] use std::os::unix::fs::MetadataExt; #[cfg(windows)] @@ -47,7 +46,7 @@ pub async fn handle( req: Request, mut res: ResponseBuilder, state: Arc, -) -> anyhow::Result>> { +) -> anyhow::Result>> { let config = state.config.clone(); let logger = state.logger.clone(); let watcher = state.watcher.clone(); diff --git a/src/platform/http1/bytes.rs b/src/platform/http1/bytes.rs index 88e7c8a..d95cc75 100644 --- a/src/platform/http1/bytes.rs +++ b/src/platform/http1/bytes.rs @@ -1,6 +1,6 @@ use std::borrow::Cow; -use std::convert::Infallible; +use http_body_util::BodyExt; use http_body_util::Full; use http_body_util::combinators::BoxBody; use hyper::body::Bytes as HyperBytes; @@ -43,8 +43,10 @@ impl From for Full { } } -impl From for BoxBody { +impl From for BoxBody { fn from(val: Bytes) -> Self { - BoxBody::new(Full::new(HyperBytes::from(val.0))) + Full::new(HyperBytes::from(val.0)) + .map_err(std::io::Error::other) + .boxed() } } diff --git a/src/platform/http1/http1_server.rs b/src/platform/http1/http1_server.rs index bbc16ec..f52fc94 100644 --- a/src/platform/http1/http1_server.rs +++ b/src/platform/http1/http1_server.rs @@ -1,7 +1,7 @@ -use std::convert::Infallible; use std::future::Future; use std::sync::Arc; +use http_body_util::BodyExt; use http_body_util::Full; use http_body_util::combinators::BoxBody; use hyper::Request; @@ -23,7 +23,7 @@ pub async fn http1_server( where A: ToSocketAddrs, F: 'static + Send + Sync + Fn(Request, ResponseBuilder) -> Fut, - Fut: Send + Future>>>, + Fut: Send + Future>>>, { let listener = TcpListener::bind(&addr).await?; let handler_func_ref = Arc::new(handle_func); @@ -46,7 +46,7 @@ where Err(handler_error) => handle_error(handler_error), }; - Ok::>, anyhow::Error>(handler_response) + Ok::>, anyhow::Error>(handler_response) } }); @@ -58,9 +58,9 @@ where } } -fn handle_error(error: anyhow::Error) -> Response> { +fn handle_error(error: anyhow::Error) -> Response> { let content = HyperBytes::from(format!("{}", error)); - let body = BoxBody::new(Full::new(content)); + let body = Full::new(content).map_err(std::io::Error::other).boxed(); let response = Response::builder().status(500).body(body); let Ok(response) = response else { todo!() }; diff --git a/src/platform/http1/res_ext.rs b/src/platform/http1/res_ext.rs index 533bd67..d1b0f9c 100644 --- a/src/platform/http1/res_ext.rs +++ b/src/platform/http1/res_ext.rs @@ -1,5 +1,3 @@ -use std::convert::Infallible; - use futures::TryStreamExt; use http::Response as HttpResponse; use http::Result as HttpResult; @@ -15,29 +13,35 @@ pub trait ResponseBuilderExt { fn body_stream( self, stream_buffer_size: usize, - ) -> HttpResult<(HttpResponse>, DuplexStream)>; + ) -> HttpResult<( + HttpResponse>, + DuplexStream, + )>; fn body_from( self, bytes: impl Into, - ) -> HttpResult>>; + ) -> HttpResult>>; } impl ResponseBuilderExt for ResponseBuilder { fn body_stream( self, stream_buffer_size: usize, - ) -> HttpResult<(HttpResponse>, DuplexStream)> { + ) -> HttpResult<( + HttpResponse>, + DuplexStream, + )> { let (writer, reader) = tokio::io::duplex(stream_buffer_size); let reader_stream = tokio_util::io::ReaderStream::new(reader) .map_ok(hyper::body::Frame::data) - .map_err(|_item| panic!()); + .map_err(std::io::Error::other); let stream_body = StreamBody::new(reader_stream); - let boxed_body: BoxBody = - BoxBody::::new(stream_body); + let boxed_body: BoxBody = + BoxBody::::new(stream_body); - let res: http::Response> = self.body(boxed_body)?; + let res: http::Response> = self.body(boxed_body)?; Ok((res, writer)) } @@ -45,7 +49,7 @@ impl ResponseBuilderExt for ResponseBuilder { fn body_from( self, bytes: impl Into, - ) -> HttpResult>> { + ) -> HttpResult>> { self.body(bytes.into().into()) } } diff --git a/src/platform/proxy/mod.rs b/src/platform/proxy/mod.rs index ec136c7..fb7d7fd 100644 --- a/src/platform/proxy/mod.rs +++ b/src/platform/proxy/mod.rs @@ -1,5 +1,4 @@ use std::collections::HashMap; -use std::convert::Infallible; use std::sync::Arc; use http_body_util::BodyExt; @@ -28,7 +27,7 @@ pub async fn proxy_request( client: &Arc, route: &ProxyRoute, req: Request, -) -> anyhow::Result>> { +) -> anyhow::Result>> { let request_path = req.uri().path().to_string(); let query = req.uri().query().map(|v| v.to_string()); let method = req.method().clone(); @@ -89,16 +88,8 @@ pub async fn proxy_request( res = res.header(key, value); } - // Collect the upstream body into memory so it can be re-framed with an - // `Infallible` error type expected by the server response body. - let body = upstream_res - .into_body() - .collect() - .await - .map_err(|err| anyhow::anyhow!("Unable to read proxied response body: {}", err))? - .to_bytes(); - - let body: BoxBody = BoxBody::new(http_body_util::Full::new(body)); + let body: BoxBody = + BoxBody::new(upstream_res.into_body().map_err(std::io::Error::other)); let res = res.body(body)?;