fetch: send streamed request body through CONNECT proxy tunnel

src/http/lib.rs

@@ -2811,6 +2811,64 @@ impl<'a> HTTPClient<'a> {
         buffer: &mut bun_io::StreamBuffer,
         data: &[u8],
     ) -> Result<bool, bun_core::Error> {
+        // When tunneling (CONNECT + inner TLS) the chunked body must go through
+        // the inner SSL wrapper; `socket` is the outer proxy connection and
+        // writing plaintext to it would corrupt the encrypted stream. The
+        // wrapper's `write_encrypted` callback pushes the ciphertext onto the
+        // outer socket.
+        if let Some(proxy_ptr) = self.proxy_tunnel.as_ref().map(|p| p.as_ptr()) {
+            // Detached upgrade so `&mut self` can be reborrowed below; the
+            // tunnel is a disjoint heap allocation (see
+            // `proxy_tunnel::raw_as_mut` INVARIANT).
+            let proxy = proxy_tunnel::raw_as_mut(proxy_ptr);
+            let to_send_len = buffer.slice().len();
+            if to_send_len > 0 {
+                match ProxyTunnel::write(proxy, buffer.slice()) {
+                    Ok(amount) => {
+                        self.state.request_sent_len += amount;
+                        buffer.cursor += amount;
+                        if amount < to_send_len {
+                            if !data.is_empty() {
+                                let _ = buffer.write(data);
+                            }
+                            return Ok(true);
+                        }
+                        if buffer.is_empty() {
+                            buffer.reset();
+                        }
+                    }
+                    Err(e) if e == err!(ConnectionClosed) => return Err(e),
+                    Err(_) => {
+                        // WantRead/WantWrite from the inner SSL: treat as
+                        // backpressure — queue any new data and retry on
+                        // the next onWritable.
+                        if !data.is_empty() {
+                            let _ = buffer.write(data);
+                        }
+                        return Ok(true);
+                    }
+                }
+            }
+            if !data.is_empty() {
+                match ProxyTunnel::write(proxy, data) {
+                    Ok(sent) => {
+                        self.state.request_sent_len += sent;
+                        if sent < data.len() {
+                            let _ = buffer.write(&data[sent..]);
+                            return Ok(true);
+                        }
+                        return Ok(false);
+                    }
+                    Err(e) if e == err!(ConnectionClosed) => return Err(e),
+                    Err(_) => {
+                        let _ = buffer.write(data);
+                        return Ok(true);
+                    }
+                }
+            }
+            return Ok(false);
+        }
+
         let to_send_len = buffer.slice().len();
         if to_send_len > 0 {
             let amount = write_to_socket::<IS_SSL>(socket, buffer.slice())?;
@@ -3176,7 +3234,12 @@ impl<'a> HTTPClient<'a> {
                         );
                     }
 
-                    let has_sent_body = self.request_body().is_empty();
+                    let has_sent_body =
+                        if matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_)) {
+                            self.request_body().is_empty()
+                        } else {
+                            false
+                        };
 
                     if has_sent_headers && has_sent_body {
                         self.state.request_stage = RequestStage::Done;
@@ -3190,7 +3253,15 @@ impl<'a> HTTPClient<'a> {
                             let ctx = self.get_ssl_ctx::<IS_SSL>();
                             self.progress_update::<IS_SSL>(ctx, socket);
                         }
-                        debug_assert!(!self.request_body().is_empty());
+                        debug_assert!(
+                            // we should have leftover data OR we use sendfile/stream
+                            (matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_))
+                                && !self.request_body().is_empty())
+                                || matches!(
+                                    self.state.original_request_body,
+                                    HTTPRequestBody::Sendfile(_) | HTTPRequestBody::Stream(_)
+                                )
+                        );
 
                         // we sent everything, but there's some body leftover
                         if amount == to_send.len() {

test/js/bun/http/proxy.test.ts

@@ -228,6 +228,52 @@ for (const proxy_tls of [false, true]) {
   }
 }
 
+// Streamed request bodies through a CONNECT tunnel: after the inner TLS
+// handshake completes and the request headers are written, the body must be
+// written through the tunnel's inner TLS wrapper (not the outer proxy socket)
+// and the ProxyHeaders stage must not short-circuit to Done just because the
+// synchronous `request_body` slice is empty for the Stream variant. Before
+// the fix both happened, so the origin waited forever for a body that was
+// never sent.
+describe("streamed request body through CONNECT tunnel", () => {
+  const streamedBodies = {
+    ReadableStream: () =>
+      new ReadableStream({
+        start(c) {
+          c.enqueue(new TextEncoder().encode("hello, "));
+          c.enqueue(new TextEncoder().encode("tunneled "));
+          c.enqueue(new TextEncoder().encode("world"));
+          c.close();
+        },
+      }),
+    "async iterator": () =>
+      (async function* () {
+        yield new TextEncoder().encode("hello, ");
+        yield new TextEncoder().encode("tunneled ");
+        yield new TextEncoder().encode("world");
+      })(),
+  } as const;
+
+  for (const proxy_tls of [false, true]) {
+    for (const [kind, makeBody] of Object.entries(streamedBodies)) {
+      test(`${kind} body through ${proxy_tls ? "TLS" : "non-TLS"} proxy -> TLS target`, async () => {
+        const response = await fetch(httpsServer.url, {
+          method: "POST",
+          proxy: proxy_tls ? httpsProxyServer.url : httpProxyServer.url,
+          headers: { "Content-Type": "text/plain" },
+          keepalive: false,
+          body: makeBody() as any,
+          duplex: "half",
+          tls: { ca: tlsCert.cert, rejectUnauthorized: false },
+        });
+        const result = await response.text();
+        expect(result).toBe("hello, tunneled world");
+        expect(response.status).toBe(200);
+      });
+    }
+  }
+});
+
 for (const server_tls of [false, true]) {
   describe.concurrent(`proxy can handle redirects with ${server_tls ? "TLS" : "non-TLS"} server`, () => {
     test("with empty body #12007", async () => {