fix: extract head bytes for CONNECT event

Add a dedicated CONNECT upgrade path that sends the 200 response via
hyper, awaits the upgrade, and captures any trailing bytes (data
pipelined after the CONNECT request headers). These bytes are passed
as the `head` parameter to the "connect" event, matching Node.js
behavior.

The new UpgradeStream `ConsumeResponse` write state absorbs the
redundant HTTP response the application writes through the socket
(since hyper already sent 200), then switches to raw network mode.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: bartlomieju

ext/http/00_serve.ts

@@ -29,6 +29,8 @@ import {
   op_http_set_response_trailers,
   op_http_try_wait,
   op_http_upgrade_raw,
+  op_http_upgrade_raw_connect,
+  op_http_upgrade_raw_get_head,
   op_http_upgrade_websocket_next,
   op_http_wait,
 } from "ext:core/ops";
@@ -151,6 +153,16 @@ function upgradeHttpRaw(req) {
   throw new TypeError("'upgradeHttpRaw' may only be used with Deno.serve");
 }
 
+function upgradeHttpRawConnect(req) {
+  const inner = toInnerRequest(req);
+  if (inner?._wantsUpgrade) {
+    return inner._wantsUpgrade("upgradeConnect");
+  }
+  throw new TypeError(
+    "'upgradeHttpRawConnect' may only be used with Deno.serve",
+  );
+}
+
 function addTrailers(resp, headerList) {
   const inner = toInnerResponse(resp);
   op_http_set_response_trailers(inner.external, headerList);
@@ -226,6 +238,31 @@ class InnerRequest {
       return { response: UPGRADE_RESPONSE_SENTINEL, conn };
     }
 
+    if (upgradeType == "upgradeConnect") {
+      const external = this.#external;
+      const remoteAddr = this.remoteAddr;
+      const localAddr = this.#context.listener.addr;
+
+      this.url();
+      this.headerList;
+      this.close();
+
+      this.#upgraded = true;
+
+      return (async () => {
+        const upgradeRid = await op_http_upgrade_raw_connect(external);
+        const head = op_http_upgrade_raw_get_head(upgradeRid);
+
+        const conn = new UpgradedConn(
+          upgradeRid,
+          remoteAddr,
+          localAddr,
+        );
+
+        return { response: UPGRADE_RESPONSE_SENTINEL, conn, head };
+      })();
+    }
+
     if (upgradeType == "upgradeWebSocket") {
       const external = this.#external;
 
@@ -1101,6 +1138,7 @@ function serveHttpOn(context, addr, callback) {
 
 internals.addTrailers = addTrailers;
 internals.upgradeHttpRaw = upgradeHttpRaw;
+internals.upgradeHttpRawConnect = upgradeHttpRawConnect;
 internals.serveHttpOnListener = serveHttpOnListener;
 internals.serveHttpOnConnection = serveHttpOnConnection;
 
@@ -1178,4 +1216,5 @@ export {
   serveHttpOnConnection,
   serveHttpOnListener,
   upgradeHttpRaw,
+  upgradeHttpRawConnect,
 };

ext/http/http_next.rs

@@ -218,6 +218,48 @@ pub fn op_http_upgrade_raw(
   Ok(state.resource_table.add(UpgradeStream::new(read, write)))
 }
 
+/// Upgrade a CONNECT request by sending a 200 response, awaiting the
+/// upgrade, and returning the stream resource with head bytes captured.
+#[op2]
+#[smi]
+pub async fn op_http_upgrade_raw_connect(
+  state: Rc<RefCell<OpState>>,
+  external: *const c_void,
+) -> Result<ResourceId, HttpNextError> {
+  let (http, upgrade) = {
+    // SAFETY: external is deleted before calling this op.
+    let http =
+      unsafe { take_external!(external, "op_http_upgrade_raw_connect") };
+    let upgrade = http.upgrade()?;
+    (http, upgrade)
+  };
+
+  // Send a 200 response to complete the CONNECT handshake.
+  http.response_parts().status = StatusCode::OK;
+  http.complete();
+
+  let upgraded = upgrade.await?;
+  let (stream, head_bytes) = extract_network_stream(upgraded);
+  let (read_half, write_half) = stream.into_split();
+
+  let resource =
+    UpgradeStream::new_connected(read_half, write_half, head_bytes);
+  Ok(state.borrow_mut().resource_table.add(resource))
+}
+
+/// Return the head bytes captured during a CONNECT upgrade.
+/// The bytes are returned exactly once; subsequent calls return empty.
+#[op2]
+#[buffer]
+pub fn op_http_upgrade_raw_get_head(
+  state: &mut OpState,
+  #[smi] rid: ResourceId,
+) -> Result<Vec<u8>, HttpNextError> {
+  let resource = state.resource_table.get::<UpgradeStream>(rid)?;
+  let bytes = resource.head_bytes.borrow_mut().take();
+  Ok(bytes.map(|b| b.to_vec()).unwrap_or_default())
+}
+
 #[op2]
 #[smi]
 pub async fn op_http_upgrade_websocket_next(
@@ -1367,6 +1409,10 @@ enum UpgradeStreamWriteState {
     OnUpgrade,
     AsyncMut<Option<(NetworkStreamReadHalf, Bytes)>>,
   ),
+  /// Used after a CONNECT upgrade where the 200 response was already sent
+  /// by hyper. Consumes and discards the HTTP response the application
+  /// writes (since it's redundant), then switches to Network mode.
+  ConsumeResponse(BytesMut, NetworkStreamWriteHalf),
   Network(NetworkStreamWriteHalf),
   Failed,
 }
@@ -1375,6 +1421,9 @@ struct UpgradeStream {
   read: Rc<AsyncRefCell<Option<(NetworkStreamReadHalf, Bytes)>>>,
   write: AsyncRefCell<UpgradeStreamWriteState>,
   cancel_handle: CancelHandle,
+  /// Head bytes extracted during a CONNECT upgrade, available via
+  /// `op_http_upgrade_raw_get_head`.
+  head_bytes: RefCell<Option<Bytes>>,
 }
 
 impl UpgradeStream {
@@ -1386,6 +1435,24 @@ impl UpgradeStream {
       read,
       write: AsyncRefCell::new(write),
       cancel_handle: CancelHandle::new(),
+      head_bytes: RefCell::new(None),
+    }
+  }
+
+  pub fn new_connected(
+    read_half: NetworkStreamReadHalf,
+    write_half: NetworkStreamWriteHalf,
+    head_bytes: Bytes,
+  ) -> Self {
+    let read = Rc::new(AsyncRefCell::new(Some((read_half, Bytes::new()))));
+    Self {
+      read,
+      write: AsyncRefCell::new(UpgradeStreamWriteState::ConsumeResponse(
+        BytesMut::with_capacity(128),
+        write_half,
+      )),
+      cancel_handle: CancelHandle::new(),
+      head_bytes: RefCell::new(Some(head_bytes)),
     }
   }
 
@@ -1478,6 +1545,44 @@ impl UpgradeStream {
             Err(e) => Err(std::io::Error::other(e)),
           }
         }
+        UpgradeStreamWriteState::ConsumeResponse(mut bytes, mut stream) => {
+          bytes.extend_from_slice(buf);
+
+          let mut headers = [httparse::EMPTY_HEADER; 16];
+          let mut response = httparse::Response::new(&mut headers);
+          match response.parse(&bytes) {
+            Ok(httparse::Status::Partial) => {
+              *wr = UpgradeStreamWriteState::ConsumeResponse(bytes, stream);
+              Ok(buf.len())
+            }
+            Ok(httparse::Status::Complete(n)) => {
+              // Response consumed. Forward any trailing bytes after
+              // the response headers to the network.
+              let trailing = &bytes[n..];
+              if !trailing.is_empty() {
+                let mut written = 0;
+                while written < trailing.len() {
+                  written +=
+                    Pin::new(&mut stream).write(&trailing[written..]).await?;
+                }
+              }
+              let consumed_from_buf = n - (bytes.len() - buf.len());
+              *wr = UpgradeStreamWriteState::Network(stream);
+              Ok(consumed_from_buf)
+            }
+            Err(_) => {
+              // Not an HTTP response — treat as raw data.
+              // Write everything accumulated so far to the network.
+              let all = bytes.freeze();
+              let mut written = 0;
+              while written < all.len() {
+                written += Pin::new(&mut stream).write(&all[written..]).await?;
+              }
+              *wr = UpgradeStreamWriteState::Network(stream);
+              Ok(buf.len())
+            }
+          }
+        }
         UpgradeStreamWriteState::Network(mut stream) => {
           let r = Pin::new(&mut stream).write(buf).await;
           *wr = UpgradeStreamWriteState::Network(stream);
@@ -1503,6 +1608,9 @@ impl UpgradeStream {
       UpgradeStreamWriteState::Parsing(..) => {
         self.write(if buf1.is_empty() { buf2 } else { buf1 }).await
       }
+      UpgradeStreamWriteState::ConsumeResponse(..) => {
+        self.write(if buf1.is_empty() { buf2 } else { buf1 }).await
+      }
       UpgradeStreamWriteState::Network(stream) => {
         let bufs = [std::io::IoSlice::new(buf1), std::io::IoSlice::new(buf2)];
         stream.write_vectored(&bufs).await

ext/http/lib.rs

@@ -171,6 +171,8 @@ deno_core::extension!(
     http_next::op_http_set_response_trailers,
     http_next::op_http_upgrade_websocket_next,
     http_next::op_http_upgrade_raw,
+    http_next::op_http_upgrade_raw_connect,
+    http_next::op_http_upgrade_raw_get_head,
     http_next::op_raw_write_vectored,
     http_next::op_can_write_vectored,
     http_next::op_http_try_wait,
@@ -221,6 +223,8 @@ deno_core::extension!(
     http_next::op_http_set_response_trailers,
     http_next::op_http_upgrade_websocket_next,
     http_next::op_http_upgrade_raw,
+    http_next::op_http_upgrade_raw_connect,
+    http_next::op_http_upgrade_raw_get_head,
     http_next::op_raw_write_vectored,
     http_next::op_can_write_vectored,
     http_next::op_http_try_wait,

ext/node/polyfills/http.ts

@@ -70,7 +70,11 @@ import {
 } from "ext:deno_node/internal/errors.ts";
 import { getTimerDuration } from "ext:deno_node/internal/timers.mjs";
 import { getIPFamily } from "ext:deno_node/internal/net.ts";
-import { serveHttpOnListener, upgradeHttpRaw } from "ext:deno_http/00_serve.ts";
+import {
+  serveHttpOnListener,
+  upgradeHttpRaw,
+  upgradeHttpRawConnect,
+} from "ext:deno_http/00_serve.ts";
 import { listen as listenDeno } from "ext:deno_net/01_net.js";
 import { headersEntries } from "ext:deno_fetch/20_headers.js";
 import { Response } from "ext:deno_fetch/23_response.js";
@@ -2283,13 +2287,17 @@ export class ServerImpl extends EventEmitter {
         req[kRawHeaders] = request.headers;
 
         if (this.listenerCount("connect") > 0) {
-          const { conn, response } = upgradeHttpRaw(request);
-          const socket = new Socket({
-            handle: new TCP(constants.SERVER, conn),
-          });
-          req.socket = socket;
-          this.emit("connect", req, socket, Buffer.from([]));
-          return response;
+          return (async () => {
+            const { conn, response, head } = await upgradeHttpRawConnect(
+              request,
+            );
+            const socket = new Socket({
+              handle: new TCP(constants.SERVER, conn),
+            });
+            req.socket = socket;
+            this.emit("connect", req, socket, Buffer.from(head));
+            return response;
+          })();
         } else {
           return new Response(null, { status: 405 });
         }