Fix HTTP/2 requests failing with duplicate Content-Length (#391)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

crates/yaak-http/src/sender.rs

@@ -2,7 +2,9 @@ use crate::decompress::{ContentEncoding, streaming_decoder};
 use crate::error::{Error, Result};
 use crate::types::{SendableBody, SendableHttpRequest};
 use async_trait::async_trait;
+use bytes::Bytes;
 use futures_util::StreamExt;
+use http_body::{Body as HttpBody, Frame, SizeHint};
 use reqwest::{Client, Method, Version};
 use std::fmt::Display;
 use std::pin::Pin;
@@ -413,10 +415,16 @@ impl HttpSender for ReqwestSender {
             Some(SendableBody::Bytes(bytes)) => {
                 req_builder = req_builder.body(bytes);
             }
-            Some(SendableBody::Stream(stream)) => {
-                // Convert AsyncRead stream to reqwest Body
-                let stream = tokio_util::io::ReaderStream::new(stream);
-                let body = reqwest::Body::wrap_stream(stream);
+            Some(SendableBody::Stream { data, content_length }) => {
+                // Convert AsyncRead stream to reqwest Body. If content length is
+                // known, wrap with a SizedBody so hyper can set Content-Length
+                // automatically (for both HTTP/1.1 and HTTP/2).
+                let stream = tokio_util::io::ReaderStream::new(data);
+                let body = if let Some(len) = content_length {
+                    reqwest::Body::wrap(SizedBody::new(stream, len))
+                } else {
+                    reqwest::Body::wrap_stream(stream)
+                };
                 req_builder = req_builder.body(body);
             }
         }
@@ -520,6 +528,51 @@ impl HttpSender for ReqwestSender {
     }
 }
 
+/// A wrapper around a byte stream that reports a known content length via
+/// `size_hint()`. This lets hyper set the `Content-Length` header
+/// automatically based on the body size, without us having to add it as an
+/// explicit header — which can cause duplicate `Content-Length` headers and
+/// break HTTP/2.
+struct SizedBody<S> {
+    stream: std::sync::Mutex<S>,
+    remaining: u64,
+}
+
+impl<S> SizedBody<S> {
+    fn new(stream: S, content_length: u64) -> Self {
+        Self { stream: std::sync::Mutex::new(stream), remaining: content_length }
+    }
+}
+
+impl<S> HttpBody for SizedBody<S>
+where
+    S: futures_util::Stream<Item = std::result::Result<Bytes, std::io::Error>> + Send + Unpin + 'static,
+{
+    type Data = Bytes;
+    type Error = std::io::Error;
+
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<std::result::Result<Frame<Self::Data>, Self::Error>>> {
+        let this = self.get_mut();
+        let mut stream = this.stream.lock().unwrap();
+        match stream.poll_next_unpin(cx) {
+            Poll::Ready(Some(Ok(chunk))) => {
+                this.remaining = this.remaining.saturating_sub(chunk.len() as u64);
+                Poll::Ready(Some(Ok(Frame::data(chunk))))
+            }
+            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+
+    fn size_hint(&self) -> SizeHint {
+        SizeHint::with_exact(self.remaining)
+    }
+}
+
 fn version_to_str(version: &Version) -> String {
     match *version {
         Version::HTTP_09 => "HTTP/0.9".to_string(),