fix(http): address streaming review findings

CLAUDE.md

@@ -341,6 +341,8 @@ When the user asks to track something in a note, store it in `~/.agents/notes/`
 - Follow existing patterns in neighboring files.
 - Always check existing imports and dependencies before adding new ones.
 - **Always add imports at the top of the file instead of inline within a function.**
+- Prefer explicit `if` or `match` statements over calling `.then(...)` on booleans.
+- Warning and error logs should briefly identify the likely operational cause or direction when known.
 
 ### Comments
 

engine/packages/config/src/config/pegboard.rs

@@ -121,6 +121,12 @@ pub struct Pegboard {
 	pub gateway_hws_max_pending_size: Option<u64>,
 	/// Max HTTP request body size in bytes for requests to actors.
 	pub gateway_http_max_request_body_size: Option<usize>,
+	/// Number of body chunks buffered between a streaming response handler and its HTTP client.
+	pub gateway_http_response_body_channel_capacity: Option<usize>,
+	/// Maximum number of envoy response messages buffered per HTTP request.
+	pub gateway_http_response_queue_max_messages: Option<usize>,
+	/// Maximum streaming response bytes buffered per HTTP request.
+	pub gateway_streaming_http_response_queue_max_bytes: Option<usize>,
 
 	// === Envoy Settings ===
 	/// How long to wait before considering an envoy lost and evicting all of its actors.
@@ -309,6 +315,23 @@ impl Pegboard {
 			.unwrap_or(128 * 1024 * 1024) // 128 MiB
 	}
 
+	pub fn gateway_http_response_body_channel_capacity(&self) -> usize {
+		self.gateway_http_response_body_channel_capacity
+			.unwrap_or(16)
+			.max(1)
+	}
+
+	pub fn gateway_http_response_queue_max_messages(&self) -> usize {
+		// A 20 MiB response occupies 320 64 KiB chunks. Leave room for
+		// control frames and smaller chunks while still bounding amplification.
+		self.gateway_http_response_queue_max_messages.unwrap_or(384)
+	}
+
+	pub fn gateway_streaming_http_response_queue_max_bytes(&self) -> usize {
+		self.gateway_streaming_http_response_queue_max_bytes
+			.unwrap_or(20 * 1024 * 1024) // 20 MiB
+	}
+
 	pub fn runner_max_response_payload_body_size(&self) -> usize {
 		self.runner_max_response_payload_body_size
 			.unwrap_or(20 * 1024 * 1024) // 20 MiB

engine/packages/guard-core/src/response_body.rs

@@ -46,6 +46,7 @@ pub enum ResponseBody {
 
 impl ResponseBody {
 	#[doc(hidden)]
+	/// Runs `callback` exactly once when the body reaches EOF, errors, or is dropped.
 	pub fn with_completion(self, callback: impl FnOnce() + Send + 'static) -> Self {
 		Self::WithCompletion {
 			body: Box::new(self),

engine/packages/pegboard-gateway2/src/http_stream/handler.rs

@@ -22,7 +22,7 @@ use tokio::sync::{mpsc, watch};
 use tracing::Instrument;
 
 use super::{
-	HTTP_RESPONSE_BODY_CHANNEL_CAPACITY, ResponseBodyError,
+	ResponseBodyError,
 	request::{
 		should_stream_http_request_body_hint, stream_http_request_and_wait_for_response,
 		wait_for_http_response_start,
@@ -293,9 +293,13 @@ impl PegboardGateway2 {
 			}
 
 			let response = if response_start.stream {
-				let (body_tx, body_rx) = mpsc::channel::<Result<Bytes, ResponseBodyError>>(
-					HTTP_RESPONSE_BODY_CHANNEL_CAPACITY,
-				);
+				let body_channel_capacity = self
+					.ctx
+					.config()
+					.pegboard()
+					.gateway_http_response_body_channel_capacity();
+				let (body_tx, body_rx) =
+					mpsc::channel::<Result<Bytes, ResponseBodyError>>(body_channel_capacity);
 				let idle_timeout = self
 					.ctx
 					.config()

engine/packages/pegboard-gateway2/src/http_stream/mod.rs

@@ -11,7 +11,6 @@ pub(crate) use response_queue::{
 	HttpResponseQueueBudget, HttpResponseQueueOverloaded, HttpResponseQueuePermit,
 };
 
-pub(super) const HTTP_RESPONSE_BODY_CHANNEL_CAPACITY: usize = 16;
 pub(super) type ResponseBodyError = Box<dyn std::error::Error + Send + Sync>;
 
 pub(super) async fn send_http_request_abort(

engine/packages/pegboard-gateway2/src/http_stream/request.rs

@@ -24,13 +24,23 @@ const HTTP_BODY_CHUNK_SIZE: usize = 64 * 1024;
 const HTTP_BODY_CHUNK_FLUSH_INTERVAL: Duration = Duration::from_millis(10);
 
 pub(super) fn should_stream_http_request_body_hint(size_hint: &SizeHint) -> bool {
-	size_hint
-		.upper()
-		.map_or(true, |body_len| body_len as usize > HTTP_BODY_CHUNK_SIZE)
+	match size_hint.upper() {
+		Some(body_len) => body_len as usize > HTTP_BODY_CHUNK_SIZE,
+		None => true,
+	}
+}
+
+#[derive(Debug, PartialEq)]
+enum RequestBodySize {
+	WithinLimit(usize),
+	ExceedsLimit,
 }
 
-fn next_request_body_size(current: usize, chunk: usize, limit: usize) -> Option<usize> {
-	current.checked_add(chunk).filter(|size| *size <= limit)
+fn next_request_body_size(current: usize, chunk: usize, limit: usize) -> RequestBodySize {
+	match current.checked_add(chunk) {
+		Some(size) if size <= limit => RequestBodySize::WithinLimit(size),
+		Some(_) | None => RequestBodySize::ExceedsLimit,
+	}
 }
 
 #[derive(Default)]
@@ -106,12 +116,17 @@ where
 		} else {
 			body.frame().await
 		};
-		let Some(frame) = frame else {
-			break;
-		};
-		let frame = match frame {
-			Ok(frame) => frame,
-			Err(error) => {
+		let data = match frame {
+			// The client request body reached normal EOF.
+			None => break,
+			Some(Ok(frame)) => {
+				let Ok(data) = frame.into_data() else {
+					continue;
+				};
+				data
+			}
+			Some(Err(error)) => {
+				tracing::warn!(%error, "failed to read streaming request body from client");
 				super::send_http_request_abort(
 					in_flight_req,
 					protocol::HttpStreamAbortReasonKind::ClientDisconnect,
@@ -121,26 +136,20 @@ where
 				return Err(anyhow!("failed to read streaming request body: {error}"));
 			}
 		};
-		let Ok(data) = frame.into_data() else {
-			continue;
-		};
 		ingress_bytes.fetch_add(data.len() as u64, Ordering::AcqRel);
-		let Some(next_body_size) = next_request_body_size(body_size, data.len(), max_body_size)
-		else {
-			super::send_http_request_abort(
-				in_flight_req,
-				protocol::HttpStreamAbortReasonKind::BodyTooLarge,
-				Some(format!(
-					"request body exceeded the {max_body_size}-byte limit"
-				)),
-			)
-			.await;
-			return Err(InvalidRequestBody {
-				reason: format!("request body exceeded the {max_body_size}-byte limit"),
+		body_size = match next_request_body_size(body_size, data.len(), max_body_size) {
+			RequestBodySize::WithinLimit(next_body_size) => next_body_size,
+			RequestBodySize::ExceedsLimit => {
+				let reason = format!("request body exceeded the {max_body_size}-byte limit");
+				super::send_http_request_abort(
+					in_flight_req,
+					protocol::HttpStreamAbortReasonKind::BodyTooLarge,
+					Some(reason.clone()),
+				)
+				.await;
+				return Err(InvalidRequestBody { reason }.build());
 			}
-			.build());
 		};
-		body_size = next_body_size;
 
 		let was_empty = chunker.is_empty();
 		for chunk in chunker.push(&data) {
@@ -153,12 +162,10 @@ where
 		}
 	}
 
-	// Normal EOF flushes any partial protocol chunk, then sends exactly one final
-	// marker so actor-side upload state and request routing can be released.
-	if let Some(chunk) = chunker.flush() {
-		send_http_request_body_chunk(in_flight_req, chunk, false).await?;
-	}
-	send_http_request_body_chunk(in_flight_req, Vec::new(), true).await
+	// Normal EOF sends exactly one final protocol chunk so actor-side upload
+	// state and request routing can be released.
+	let final_chunk = chunker.flush().unwrap_or_default();
+	send_http_request_body_chunk(in_flight_req, final_chunk, true).await
 }
 
 async fn send_http_request_body_chunk(
@@ -267,35 +274,28 @@ where
 	let upload =
 		send_streaming_http_request_body_chunks(in_flight_req, body, max_body_size, ingress_bytes);
 	tokio::pin!(upload);
+	let response_start = wait_for_http_response_start(
+		msg_rx,
+		drop_rx,
+		stopped_sub,
+		actor_id,
+		request_id,
+		response_start_deadline,
+		response_start_timeout,
+	);
+	tokio::pin!(response_start);
 	tokio::select! {
 		upload_result = &mut upload => {
 			// Normal completion already sent the final upload marker. Continue
-			// waiting under the original response-start deadline.
+			// polling the same response future under the original deadline.
 			upload_result?;
-			wait_for_http_response_start(
-				msg_rx,
-				drop_rx,
-				stopped_sub,
-				actor_id,
-				request_id,
-				response_start_deadline,
-				response_start_timeout,
-			)
-			.await
+			response_start.await
 		}
-		response_start = wait_for_http_response_start(
-			msg_rx,
-			drop_rx,
-			stopped_sub,
-			actor_id,
-			request_id,
-			response_start_deadline,
-			response_start_timeout,
-		) => {
+		response_start = &mut response_start => {
 			let response_start = response_start?;
-			// An early successful response makes the unread client body
-			// irrelevant. Drop the upload future and send a clean final marker
-			// so actor request routing can be released.
+			// HTTP handlers may intentionally respond before consuming the full
+			// upload. Stop reading the client and send a clean final marker so
+			// actor request routing can be released.
 			send_http_request_body_chunk(in_flight_req, Vec::new(), true).await?;
 			Ok(response_start)
 		}

engine/packages/pegboard-gateway2/src/http_stream/response.rs

@@ -15,7 +15,10 @@ use crate::shared_state::{
 use super::{ResponseBodyError, send_http_request_abort};
 
 const HTTP_BODY_CHUNK_SIZE: usize = 64 * 1024;
+const HTTP_RESPONSE_QUEUE_OVERLOADED_DETAIL: &str =
+	"actor response producer outpaced downstream delivery and filled the gateway buffer";
 
+/// Advances the expected tunnel sequence while rejecting missing or reordered response frames.
 fn advance_http_stream_message_index(
 	expected: protocol::MessageIndex,
 	actual: protocol::MessageIndex,
@@ -54,7 +57,7 @@ async fn send_http_response_body_bytes(
 					send_http_request_abort(
 						in_flight_req,
 						protocol::HttpStreamAbortReasonKind::Overloaded,
-						Some("gateway streaming response queue overloaded".to_owned()),
+						Some(HTTP_RESPONSE_QUEUE_OVERLOADED_DETAIL.to_owned()),
 					)
 					.await;
 				}
@@ -160,17 +163,21 @@ pub(super) async fn drain_http_response_stream(
 
 				match msg.message_kind {
 					protocol::ToRivetTunnelMessageKind::ToRivetResponseChunk(chunk) => {
-						if !chunk.body.is_empty() && !send_http_response_body_bytes(
-							&in_flight_req,
-							&body_tx,
-							&mut drop_rx,
-							&mut stopped_sub,
-							actor_id,
-							chunk.body,
-							"client dropped streaming response body",
-							&egress_bytes,
-						).await {
-							return;
+						if !chunk.body.is_empty() {
+							let delivered = send_http_response_body_bytes(
+								&in_flight_req,
+								&body_tx,
+								&mut drop_rx,
+								&mut stopped_sub,
+								actor_id,
+								chunk.body,
+								"client dropped streaming response body",
+								&egress_bytes,
+							)
+							.await;
+							if !delivered {
+								return;
+							}
 						}
 
 						if chunk.finish {
@@ -223,7 +230,7 @@ pub(super) async fn drain_http_response_stream(
 					send_http_request_abort(
 						&in_flight_req,
 						protocol::HttpStreamAbortReasonKind::Overloaded,
-						Some("gateway streaming response queue overloaded".to_owned()),
+						Some(HTTP_RESPONSE_QUEUE_OVERLOADED_DETAIL.to_owned()),
 					)
 					.await;
 				}
@@ -277,7 +284,16 @@ fn send_http_body_error(
 	body_tx: &mpsc::Sender<Result<Bytes, ResponseBodyError>>,
 	message: impl Into<String>,
 ) {
-	let _ = body_tx.try_send(Err(Box::new(std::io::Error::other(message.into()))));
+	let error: Result<Bytes, ResponseBodyError> =
+		Err(Box::new(std::io::Error::other(message.into())));
+	match body_tx.try_send(error) {
+		Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => {}
+		Err(mpsc::error::TrySendError::Full(_)) => {
+			tracing::warn!(
+				"could not deliver streaming response error because the downstream buffer is full"
+			);
+		}
+	}
 }
 
 #[cfg(test)]