perf(client): stream proxy response body instead of buffering

Algorithm5838 · Algorithm5838 · commit 86e19fe100be · 2026-04-30T09:52:15.000+03:00
diff --git a/src/client.rs b/src/client.rs
@@ -1,8 +1,10 @@
 use arc_swap::ArcSwap;
 use cached::proc_macro::cached;
+use bytes::Bytes;
 use futures_lite::future::block_on;
-use futures_lite::{future::Boxed, FutureExt};
-use http_body_util::BodyExt;
+use futures_lite::{future::Boxed, FutureExt, StreamExt};
+use http_body_util::{BodyExt, StreamBody};
+use hyper::body::Frame;
 use hyper::{header, Method, Request, Response};
 use log::{error, info, trace, warn};
 use percent_encoding::{percent_encode, CONTROLS};
@@ -18,7 +20,7 @@ use std::result::Result;
 
 use crate::dbg_msg;
 use crate::oauth::{force_refresh_token, token_daemon, Oauth, OauthBackendImpl};
-use crate::server::RequestExt;
+use crate::server::{Body, RequestExt};
 use crate::utils::{format_url, Post};
 
 const REDDIT_URL_BASE: &str = "https://oauth.reddit.com";
@@ -69,8 +71,6 @@ pub fn build_client() -> WreqClient {
 		.expect("Should always be able to build a client")
 }
 
-use crate::server::{full, Body};
-
 /// Convert a wreq Response into a hyper Response<Body>.
 /// wreq and hyper now both use http v1.x, so header types are directly compatible.
 trait IntoHyperResponse {
@@ -93,15 +93,18 @@ impl IntoHyperResponse for wreq::Response {
 			}
 		}
 
-		let bytes = self.bytes().await.map_err(|e| e.to_string())?;
-		// wreq already decompresses; remove the content-encoding header so
-		// downstream code doesn't try to decompress again.
+		// wreq already decompresses; drop content-length since the final
+		// size is unknown without buffering, relying on chunked encoding.
 		if let Some(h) = builder.headers_mut() {
 			h.remove(header::CONTENT_ENCODING);
-			h.insert(header::CONTENT_LENGTH, bytes.len().into());
+			h.remove(header::CONTENT_LENGTH);
 		}
 
-		builder.body(full(bytes)).map_err(|e| e.to_string())
+		let stream = self
+			.bytes_stream()
+			.map(|r| r.map(Frame::data).map_err(|e| e.to_string()));
+
+		builder.body(BodyExt::boxed(StreamBody::new(stream))).map_err(|e| e.to_string())
 	}
 }
 
diff --git a/src/server.rs b/src/server.rs
@@ -29,17 +29,19 @@ use tokio::net::TcpListener;
 
 use crate::dbg_msg;
 
-/// The unified body type used for all responses.
-pub type Body = BoxBody<Bytes, Infallible>;
+/// Unified response body type. Error is `String` so proxy streams can
+/// propagate upstream failures; `full()`/`empty()` are backed by
+/// `Full`/`Empty<_, Infallible>` and can never produce one.
+pub type Body = BoxBody<Bytes, String>;
 
 /// Create a response body from a string or bytes.
 pub fn full<T: Into<Bytes>>(chunk: T) -> Body {
-	Full::new(chunk.into()).boxed()
+	Full::new(chunk.into()).map_err(|e: Infallible| match e {}).boxed()
 }
 
 /// Create an empty response body.
 pub fn empty() -> Body {
-	Empty::<Bytes>::new().boxed()
+	Empty::<Bytes>::new().map_err(|e: Infallible| match e {}).boxed()
 }
 
 const BANNED_USER_AGENTS: &[&str] = &[
@@ -604,8 +606,8 @@ async fn compress_response(req_headers: &HeaderMap<header::HeaderValue>, res: &m
 	let body_bytes: Vec<u8> = {
 		// Swap body with empty, collect old body
 		let old_body = std::mem::replace(res.body_mut(), empty());
-		// Body error type is Infallible, so collect cannot fail
-		old_body.collect().await.expect("Body<Infallible> collect cannot fail").to_bytes().to_vec()
+		// full/empty bodies cannot produce a String error; only proxy stream bodies can.
+		old_body.collect().await.expect("static response body collection cannot fail").to_bytes().to_vec()
 	};
 
 	// Don't bother compressing tiny responses
diff --git a/src/settings.rs b/src/settings.rs
@@ -71,7 +71,7 @@ pub async fn set(req: Request<Body>) -> Result<Response<Body>, String> {
 		.filter_map(|header| Cookie::parse(header.to_str().unwrap_or_default()).ok())
 		.collect();
 
-	// Collect the body bytes (Body error is Infallible, so unwrap is safe)
+	// Request body is pre-buffered as Full<Bytes>; the String error cannot fire.
 	let body_bytes = body.collect().await.unwrap().to_bytes();
 
 	let form = url::form_urlencoded::parse(&body_bytes).collect::<HashMap<_, _>>();
@@ -263,7 +263,7 @@ pub async fn update(req: Request<Body>) -> Result<Response<Body>, String> {
 }
 
 pub async fn encoded_restore(req: Request<Body>) -> Result<Response<Body>, String> {
-	// Body error is Infallible, so unwrap is safe
+	// Request body is pre-buffered as Full<Bytes>; the String error cannot fire.
 	let body = req.into_body().collect().await.unwrap().to_bytes();
 
 	if body.len() > 1024 * 1024 {
