fix(proxy): stream proxied response bodies instead of buffering

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<Incoming>,
   mut res: ResponseBuilder,
   state: Arc<ServeState>,
-) -> anyhow::Result<Response<BoxBody<HyperBytes, Infallible>>> {
+) -> anyhow::Result<Response<BoxBody<HyperBytes, std::io::Error>>> {
   let config = state.config.clone();
   let logger = state.logger.clone();
   let watcher = state.watcher.clone();

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<Bytes> for Full<HyperBytes> {
   }
 }
 
-impl From<Bytes> for BoxBody<HyperBytes, Infallible> {
+impl From<Bytes> for BoxBody<HyperBytes, std::io::Error> {
   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()
   }
 }

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<F, Fut, A>(
 where
   A: ToSocketAddrs,
   F: 'static + Send + Sync + Fn(Request<Incoming>, ResponseBuilder) -> Fut,
-  Fut: Send + Future<Output = anyhow::Result<Response<BoxBody<HyperBytes, Infallible>>>>,
+  Fut: Send + Future<Output = anyhow::Result<Response<BoxBody<HyperBytes, std::io::Error>>>>,
 {
   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::<Response<BoxBody<HyperBytes, Infallible>>, anyhow::Error>(handler_response)
+          Ok::<Response<BoxBody<HyperBytes, std::io::Error>>, anyhow::Error>(handler_response)
         }
       });
 
@@ -58,9 +58,9 @@ where
   }
 }
 
-fn handle_error(error: anyhow::Error) -> Response<BoxBody<HyperBytes, Infallible>> {
+fn handle_error(error: anyhow::Error) -> Response<BoxBody<HyperBytes, std::io::Error>> {
   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!() };

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,37 +13,43 @@ pub trait ResponseBuilderExt {
   fn body_stream(
     self,
     stream_buffer_size: usize,
-  ) -> HttpResult<(HttpResponse<BoxBody<HyperBytes, Infallible>>, DuplexStream)>;
+  ) -> HttpResult<(
+    HttpResponse<BoxBody<HyperBytes, std::io::Error>>,
+    DuplexStream,
+  )>;
   fn body_from(
     self,
     bytes: impl Into<Bytes>,
-  ) -> HttpResult<HttpResponse<BoxBody<HyperBytes, Infallible>>>;
+  ) -> HttpResult<HttpResponse<BoxBody<HyperBytes, std::io::Error>>>;
 }
 
 impl ResponseBuilderExt for ResponseBuilder {
   fn body_stream(
     self,
     stream_buffer_size: usize,
-  ) -> HttpResult<(HttpResponse<BoxBody<HyperBytes, Infallible>>, DuplexStream)> {
+  ) -> HttpResult<(
+    HttpResponse<BoxBody<HyperBytes, std::io::Error>>,
+    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<HyperBytes, Infallible> =
-      BoxBody::<HyperBytes, Infallible>::new(stream_body);
+    let boxed_body: BoxBody<HyperBytes, std::io::Error> =
+      BoxBody::<HyperBytes, std::io::Error>::new(stream_body);
 
-    let res: http::Response<BoxBody<HyperBytes, Infallible>> = self.body(boxed_body)?;
+    let res: http::Response<BoxBody<HyperBytes, std::io::Error>> = self.body(boxed_body)?;
 
     Ok((res, writer))
   }
 
   fn body_from(
     self,
     bytes: impl Into<Bytes>,
-  ) -> HttpResult<HttpResponse<BoxBody<HyperBytes, Infallible>>> {
+  ) -> HttpResult<HttpResponse<BoxBody<HyperBytes, std::io::Error>>> {
     self.body(bytes.into().into())
   }
 }

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<ProxyClient>,
   route: &ProxyRoute,
   req: Request<Incoming>,
-) -> anyhow::Result<Response<BoxBody<HyperBytes, Infallible>>> {
+) -> anyhow::Result<Response<BoxBody<HyperBytes, std::io::Error>>> {
   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<HyperBytes, Infallible> = BoxBody::new(http_body_util::Full::new(body));
+  let body: BoxBody<HyperBytes, std::io::Error> =
+    BoxBody::new(upstream_res.into_body().map_err(std::io::Error::other));
 
   let res = res.body(body)?;