diff --git a/changelog.d/10668-http-client-response-surface.md b/changelog.d/10668-http-client-response-surface.md new file mode 100644 index 0000000000..2f6787ba6b --- /dev/null +++ b/changelog.d/10668-http-client-response-surface.md @@ -0,0 +1,25 @@ +Fixed three `node:http`/`node:https` client-side defects. The client `IncomingMessage` now exposes +`rawHeaders`/`httpVersion`/`httpVersionMajor`/`httpVersionMinor`/`complete` (previously `undefined` on both the +typed and dynamically-dispatched surface); `httpVersion*`/`complete` fall back to the server-side accessor when +the handle is a server `IncomingMessage`, since the codegen native table shares one `class_filter` namespace +across client and server (#10467 — `rawHeaders` header-name casing on the pooled reqwest transport is a known +remaining gap, documented in the PR). `http.request`'s client now fires `req.on('upgrade', (res, socket, head) => +...)` on a `101 Switching Protocols` response instead of delivering it as an ordinary `'response'`: an upgrade +request speaks HTTP/1.1 over a raw socket (mirroring the existing trailer-aware bypass), and on `101` adopts the +stream as a `net.Socket` via `perry_ext_net::adopt_upgraded_tcp_stream` — write, inbound data delivery, and the +`head` Buffer (always a Buffer, never `undefined`, even zero-length) all match Node (#10468). The request option +`options.createConnection` (distinct from `agent.createConnection`) is now honored when the request has no +explicit Agent, taking the same raw-socket path the Agent-level override already used (#10469). + +Follow-up (unrooted-local-shape ratchet, caught before merge): `build_raw_headers_array` +(`res.rawHeaders`, added for #10467 above) held its result array's raw pointer in a plain local across +`alloc_string`/`js_array_push` calls that can allocate and therefore collect — a stale-pointer-after-collection +shape (`scripts/unrooted_local_shape.py`), not merely a scanner nit. Rooted it through +`perry_ffi::TransientRootScope::root_nanbox` and re-derive the pointer via `.get()` after each allocating call +instead of reusing the pre-call copy, matching the pattern already used throughout this crate (e.g. +`agent.rs`, `client_events.rs`). Found and fixed the same pre-existing shape in the neighboring +`set-cookie` array builder in `build_response_headers_object` (unrelated to this PR's diff, same file); extracted +it into its own top-level `build_set_cookie_array` so the rooting lines aren't deep enough for `rustfmt` to wrap a +`let` binding across lines, which had been hiding the second half of the binding from the ratchet's +line-oriented scanner. `scripts/unrooted_local_shape.py --check` now reports 558 (down from the pre-PR baseline +of 561; response_headers.rs's own ceiling drops from 1 to 0). diff --git a/crates/perry-codegen/src/lower_call/native_table/http_server.rs b/crates/perry-codegen/src/lower_call/native_table/http_server.rs index f32c3f8d47..5876c5a9f3 100644 --- a/crates/perry-codegen/src/lower_call/native_table/http_server.rs +++ b/crates/perry-codegen/src/lower_call/native_table/http_server.rs @@ -691,7 +691,12 @@ pub(super) const HTTP_SERVER_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "httpVersion", class_filter: Some("IncomingMessage"), - runtime: "js_node_http_im_http_version", + // #10467 — route through the client accessor (falls back to the + // server one internally, `server_incoming_property`) so a + // client-side `res.httpVersion` resolves instead of reading the + // server-only registry and returning the "1.1" default for every + // client response. + runtime: "js_http_response_http_version", args: &[], ret: NR_STR, }, @@ -732,12 +737,14 @@ pub(super) const HTTP_SERVER_ROWS: &[NativeModSig] = &[ args: &[], ret: NR_STR, }, + // #10467 — same client-accessor-with-server-fallback shape as the + // bare `httpVersion` entry above. NativeModSig { module: "http", has_receiver: true, method: "__get_httpVersion", class_filter: Some("IncomingMessage"), - runtime: "js_node_http_im_http_version", + runtime: "js_http_response_http_version", args: &[], ret: NR_STR, }, @@ -746,7 +753,16 @@ pub(super) const HTTP_SERVER_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "__get_httpVersionMajor", class_filter: Some("IncomingMessage"), - runtime: "js_node_http_im_http_version_major", + runtime: "js_http_response_http_version_major", + args: &[], + ret: NR_F64, + }, + NativeModSig { + module: "http", + has_receiver: true, + method: "httpVersionMajor", + class_filter: Some("IncomingMessage"), + runtime: "js_http_response_http_version_major", args: &[], ret: NR_F64, }, @@ -755,18 +771,58 @@ pub(super) const HTTP_SERVER_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "__get_httpVersionMinor", class_filter: Some("IncomingMessage"), - runtime: "js_node_http_im_http_version_minor", + runtime: "js_http_response_http_version_minor", args: &[], ret: NR_F64, }, + NativeModSig { + module: "http", + has_receiver: true, + method: "httpVersionMinor", + class_filter: Some("IncomingMessage"), + runtime: "js_http_response_http_version_minor", + args: &[], + ret: NR_F64, + }, + // `js_http_response_complete` already returns a boxed JS boolean (f64 + // bit pattern), not a raw C `i32` like the server-only accessor this + // replaced — hence `NR_F64`, matching `headers`/`trailers`/`socket` + // below (also client accessors returning pre-boxed values). NativeModSig { module: "http", has_receiver: true, method: "__get_complete", class_filter: Some("IncomingMessage"), - runtime: "js_node_http_im_complete", + runtime: "js_http_response_complete", args: &[], - ret: NR_I32, + ret: NR_F64, + }, + NativeModSig { + module: "http", + has_receiver: true, + method: "complete", + class_filter: Some("IncomingMessage"), + runtime: "js_http_response_complete", + args: &[], + ret: NR_F64, + }, + NativeModSig { + module: "http", + has_receiver: true, + method: "__get_rawHeaders", + class_filter: Some("IncomingMessage"), + runtime: "js_http_response_raw_headers", + args: &[], + ret: NR_F64, + }, + NativeModSig { + module: "http", + has_receiver: true, + method: "rawHeaders", + class_filter: Some("IncomingMessage"), + runtime: "js_http_response_raw_headers", + args: &[], + ret: NR_F64, }, NativeModSig { module: "http", diff --git a/crates/perry-codegen/src/runtime_decls/stdlib_ffi/net_http.rs b/crates/perry-codegen/src/runtime_decls/stdlib_ffi/net_http.rs index 4628dbdfec..60109cc7cb 100644 --- a/crates/perry-codegen/src/runtime_decls/stdlib_ffi/net_http.rs +++ b/crates/perry-codegen/src/runtime_decls/stdlib_ffi/net_http.rs @@ -141,6 +141,12 @@ pub(crate) fn declare_net_http(module: &mut LlModule) { ); module.declare_function("js_http_response_headers", DOUBLE, &[I64]); module.declare_function("js_http_response_trailers", DOUBLE, &[I64]); + // #10467 — client rawHeaders / httpVersion* / complete accessors. + module.declare_function("js_http_response_raw_headers", DOUBLE, &[I64]); + module.declare_function("js_http_response_http_version", I64, &[I64]); + module.declare_function("js_http_response_http_version_major", DOUBLE, &[I64]); + module.declare_function("js_http_response_http_version_minor", DOUBLE, &[I64]); + module.declare_function("js_http_response_complete", DOUBLE, &[I64]); module.declare_function("js_http_incoming_message_socket", DOUBLE, &[I64]); module.declare_function("js_http_incoming_message_req", DOUBLE, &[I64]); module.declare_function("js_http_incoming_message_set_encoding", I64, &[I64, I64]); diff --git a/crates/perry-ext-http/src/agent.rs b/crates/perry-ext-http/src/agent.rs index f53f626d7c..08307331b9 100644 --- a/crates/perry-ext-http/src/agent.rs +++ b/crates/perry-ext-http/src/agent.rs @@ -478,6 +478,15 @@ unsafe fn read_closure_field(obj_f64: f64, field: &str) -> i64 { } } +/// Extract `options.createConnection` (#10469) — the request-level socket +/// override Node honors when the caller does not pass an explicit `agent`. +/// Like `options.agent`, a closure doesn't survive the `options` JSON +/// round-trip (`parse_options_object`), so this reads the NaN-boxed field +/// straight off the original object instead. +pub(crate) unsafe fn request_create_connection_from_options(options_f64: f64) -> i64 { + read_closure_field(options_f64, "createConnection") +} + /// Extract an `options.agent` handle from `options_f64`. Returns `None` /// when the field is missing, not a pointer, or doesn't resolve to an /// AgentHandle. @@ -1675,9 +1684,43 @@ pub(crate) unsafe fn try_create_connection_socket( if cc == 0 { return None; } + invoke_create_connection_closure(cc, Some(handle), host, port, path) +} + +/// #10469 — invoke the request option's own `createConnection` override +/// (no explicit Agent involved, so there's no `AgentHandle` to pull +/// `keepAlive` defaults from — Node's own default Agent has `keepAlive: +/// false`, matched by `build_connect_options(None, ...)`). +pub(crate) unsafe fn try_request_create_connection_socket( + closure_ptr: i64, + host: &str, + port: u16, + path: &str, +) -> Option { + if closure_ptr == 0 { + return None; + } + invoke_create_connection_closure(closure_ptr, None, host, port, path) +} + +/// Shared tail of both `createConnection` invocation paths: root +/// `closure_ptr` *before* calling `build_connect_options` (it allocates — +/// without rooting first, a GC during that allocation could move the +/// closure out from under the raw `i64` copy, matching the ordering the +/// original #2154 code used), call it with `{ host, port, path, keepAlive, +/// keepAliveInitialDelay }`, and extract the `net.Socket` handle id it +/// returns. Main thread only — JS closure calls must not run on a tokio +/// worker. +unsafe fn invoke_create_connection_closure( + closure_ptr: i64, + agent_handle: Option, + host: &str, + port: u16, + path: &str, +) -> Option { let scope = perry_ffi::TransientRootScope::enter(); - let cc = scope.root_addr(cc); - let options = scope.root_nanbox(build_connect_options(handle, host, port, path)); + let cc = scope.root_addr(closure_ptr); + let options = scope.root_nanbox(build_connect_options(agent_handle, host, port, path)); let closure = JsClosure::from_raw(cc.get() as *const RawClosureHeader); let ret = closure.call1(options.get()); @@ -1711,7 +1754,7 @@ pub(crate) fn create_socket_override(handle: Handle) -> i64 { /// Returns a NaN-boxed object pointer as `f64`, or NaN-boxed `undefined` on /// allocation failure. pub(crate) unsafe fn build_connect_options( - handle: Handle, + handle: Option, host: &str, port: u16, path: &str, @@ -1750,9 +1793,12 @@ pub(crate) unsafe fn build_connect_options( 2, JsValue::from_string_ptr(path_s.as_raw()), ); - let (keep_alive, keep_alive_msecs) = agent_field(handle, (false, 1000.0), |agent| { - (agent.keep_alive, agent.keep_alive_msecs) - }); + let (keep_alive, keep_alive_msecs) = match handle { + Some(h) => agent_field(h, (false, 1000.0), |agent| { + (agent.keep_alive, agent.keep_alive_msecs) + }), + None => (false, 1000.0), + }; perry_ffi::js_object_set_field( JsValue::from_bits(obj.get().to_bits()).as_pointer(), 3, diff --git a/crates/perry-ext-http/src/client_connect_override.rs b/crates/perry-ext-http/src/client_connect_override.rs new file mode 100644 index 0000000000..bbe5bbf7e1 --- /dev/null +++ b/crates/perry-ext-http/src/client_connect_override.rs @@ -0,0 +1,211 @@ +//! Client requests routed over a caller-supplied raw socket instead of +//! reqwest: both `agent.createConnection`/`agent.createSocket` (#2154) and +//! the request option's own `createConnection` (#10469, honored only when +//! `agent_handle == 0`) end up here. Split out of `lib.rs` to stay under +//! the file-size cap; the closure storage/invocation and the `{ host, port, +//! path, keepAlive, keepAliveInitialDelay }` options object still live in +//! `agent.rs` alongside the pre-existing Agent-level override. + +use std::collections::HashMap; + +use perry_ffi::{spawn_blocking_with_reactor as spawn_blocking, Handle}; + +use super::agent; +use crate::{parse_http_response, push_event, ClientInflightGuard, PendingHttpEvent}; + +/// Look up `request_handle`'s own `createConnection` (if any) and, when +/// set, dispatch over it. `None` means "not set / not usable" — the +/// caller (only reached when `agent_handle == 0`) falls back to reqwest. +pub(crate) fn dispatch_for_handle(request_handle: Handle, url: &str) -> Option { + let cc = perry_ffi::with_handle_mut::(request_handle, |r| { + r.request_create_connection + }) + .unwrap_or(0); + if cc == 0 { + return None; + } + request_create_connection_socket(cc, url) +} + +/// The whole "no explicit Agent, but the request's own `createConnection` +/// is set" path: resolve `(host, port, path)` from `url`, invoke the +/// override on the main thread, and attach raw mode on the socket it +/// returns (so no inbound byte gets dispatched as a JS `'data'` event +/// before `dispatch_request_over_socket`'s task takes over — mirrors the +/// Agent-override path in `dispatch_request_snapshot`). `None` means "not +/// handled", so the caller falls back to the reqwest path. +pub(crate) fn request_create_connection_socket( + request_create_connection: i64, + url: &str, +) -> Option { + let (host, port, path) = super::socket_connect_target(url)?; + let socket_id = unsafe { + agent::try_request_create_connection_socket(request_create_connection, &host, port, &path) + }?; + if let Some(vt) = perry_ffi::raw_net() { + (vt.attach)(socket_id); + } + Some(socket_id) +} + +/// Serialize an HTTP/1.1 request (request line + headers + body) into the +/// bytes to write onto a socket. Forces `Connection: close` (the raw socket +/// path reads until EOF), drops any caller-supplied `Connection`/`Host` +/// header (we set `Host` from the URL), and adds `Content-Length` when a +/// body is present and the caller didn't. +fn serialize_http_request( + method: &str, + path: &str, + host_header: &str, + headers: &HashMap, + body: &[u8], +) -> Vec { + let mut req = format!("{} {} HTTP/1.1\r\nHost: {}\r\n", method, path, host_header); + let mut has_content_length = false; + for (k, v) in headers { + if k.eq_ignore_ascii_case("content-length") { + has_content_length = true; + } + if k.eq_ignore_ascii_case("connection") || k.eq_ignore_ascii_case("host") { + continue; + } + req.push_str(k); + req.push_str(": "); + req.push_str(v); + req.push_str("\r\n"); + } + req.push_str("Connection: close\r\n"); + if !body.is_empty() && !has_content_length { + req.push_str(&format!("Content-Length: {}\r\n", body.len())); + } + req.push_str("\r\n"); + let mut out = req.into_bytes(); + out.extend_from_slice(body); + out +} + +/// #2154 — run an HTTP exchange over a socket that a `createConnection` +/// override (Agent-level or, since #10469, request-level) produced +/// (`socket_id`), instead of through reqwest. Writes the serialized +/// request, reads the response until the peer closes (we force +/// `Connection: close`), parses it with [`parse_http_response`], and pushes +/// the same `Response` / `Error` event the reqwest path produces — so the +/// IncomingMessage surface is identical. +/// +/// The socket I/O goes through perry-ffi's raw-net vtable (published by +/// perry-ext-net), so this crate needs no link edge to perry-ext-net. If no +/// net backend is linked the request errors out (the override couldn't have +/// produced a socket without `net`, so this is a defensive guard). +pub(crate) fn dispatch_request_over_socket( + request_handle: Handle, + method: String, + url: String, + headers: HashMap, + body: Vec, + timeout_ms: Option, + socket_id: i64, +) { + let parsed = match reqwest::Url::parse(&url) { + Ok(u) => u, + Err(e) => { + push_event(PendingHttpEvent::Error { + request_handle, + error_message: e.to_string(), + }); + return; + } + }; + let host = parsed.host_str().unwrap_or("localhost").to_string(); + let host_header = match parsed.port() { + Some(p) => format!("{}:{}", host, p), + None => host, + }; + let mut path = parsed.path().to_string(); + if path.is_empty() { + path.push('/'); + } + if let Some(q) = parsed.query() { + path.push('?'); + path.push_str(q); + } + let req_bytes = serialize_http_request(&method, &path, &host_header, &headers, &body); + let deadline = std::time::Duration::from_millis(timeout_ms.unwrap_or(30_000)); + + spawn_blocking(move || { + let try_h = tokio::runtime::Handle::try_current(); + std::hint::black_box(&try_h); + if try_h.is_err() { + push_event(PendingHttpEvent::Error { + request_handle, + error_message: "http client runtime unavailable".to_string(), + }); + return; + } + let handle = tokio::runtime::Handle::current(); + // #5779 follow-up: keep this fetch counted in-flight for its whole + // lifetime so the idle-kick recovers a lost worker-unpark. + let inflight_guard = ClientInflightGuard::new(request_handle); + let jh = handle.spawn(async move { + let _inflight = inflight_guard; + let vtable = match perry_ffi::raw_net() { + Some(v) => v, + None => { + push_event(PendingHttpEvent::Error { + request_handle, + error_message: "agent.createConnection requires node:net (not linked)" + .to_string(), + }); + return; + } + }; + // Attach is idempotent — the request path also attaches on the + // main thread before this task runs, to close any data race. + (vtable.attach)(socket_id); + if (vtable.write)(socket_id, req_bytes.as_ptr(), req_bytes.len()) == 0 { + push_event(PendingHttpEvent::Error { + request_handle, + error_message: "failed to write request to agent socket".to_string(), + }); + return; + } + + let mut raw = Vec::new(); + let mut chunk = [0u8; 16 * 1024]; + let start = tokio::time::Instant::now(); + loop { + let n = (vtable.poll_read)(socket_id, chunk.as_mut_ptr(), chunk.len()); + if n > 0 { + raw.extend_from_slice(&chunk[..n as usize]); + } else if n == 0 { + break; // clean EOF — peer closed after the response + } else { + if start.elapsed() >= deadline { + (vtable.close)(socket_id); + push_event(PendingHttpEvent::Timeout { request_handle }); + return; + } + tokio::time::sleep(std::time::Duration::from_millis(1)).await; + } + } + (vtable.close)(socket_id); + + match parse_http_response(&raw) { + Ok(parsed) => push_event(PendingHttpEvent::Response { + request_handle, + status: parsed.status, + status_message: parsed.status_message, + headers: parsed.headers, + trailers: parsed.trailers, + body: parsed.body, + http_version: parsed.http_version, + }), + Err(error_message) => push_event(PendingHttpEvent::Error { + request_handle, + error_message, + }), + } + }); + std::hint::black_box(&jh); + std::mem::forget(jh); + }); +} diff --git a/crates/perry-ext-http/src/client_dispatch.rs b/crates/perry-ext-http/src/client_dispatch.rs index 34d418c4ab..8ce6af9ab4 100644 --- a/crates/perry-ext-http/src/client_dispatch.rs +++ b/crates/perry-ext-http/src/client_dispatch.rs @@ -20,6 +20,21 @@ use crate::{ /// fresh detached task on the same multi-thread runtime; it drives /// itself via `await` chains while we return immediately. Mirrors /// the `spawn_socket_runner` pattern in `perry-ext-net`. +/// #10467 — map a reqwest response's negotiated HTTP version to the +/// `(major, minor)` pair `IncomingMessage.httpVersion*` expects. The pooled +/// client only ever sees these five; anything else (there isn't one today) +/// falls back to `(1, 1)`. +fn reqwest_version_pair(v: reqwest::Version) -> (u8, u8) { + match v { + reqwest::Version::HTTP_09 => (0, 9), + reqwest::Version::HTTP_10 => (1, 0), + reqwest::Version::HTTP_11 => (1, 1), + reqwest::Version::HTTP_2 => (2, 0), + reqwest::Version::HTTP_3 => (3, 0), + _ => (1, 1), + } +} + pub(crate) fn dispatch_request( request_handle: Handle, method: String, @@ -79,6 +94,28 @@ pub(crate) fn dispatch_request( let inflight_guard = ClientInflightGuard::new(request_handle); let jh = handle.spawn(async move { let _inflight = inflight_guard; + // #10468 — `Connection: Upgrade` needs the raw socket handed + // back on `101`, which reqwest can't do. Checked before the + // trailer-aware bypass below (disjoint triggers: `TE: trailers` + // vs `Connection: Upgrade`, never both on the same request). + if let Some(result) = crate::client_upgrade::dispatch_upgrade_http_request( + request_handle, + method.as_str(), + &url, + &headers, + &body, + timeout_ms, + ) + .await + { + if let Err(error_message) = result { + push_event(PendingHttpEvent::Error { + request_handle, + error_message, + }); + } + return; + } if let Some(result) = dispatch_plain_http_request( request_handle, method.as_str(), @@ -148,6 +185,7 @@ pub(crate) fn dispatch_request( .canonical_reason() .unwrap_or("") .to_string(); + let http_version = reqwest_version_pair(response.version()); let mut hdrs = Vec::new(); for (k, v) in response.headers() { if let Ok(s) = v.to_str() { @@ -164,6 +202,7 @@ pub(crate) fn dispatch_request( status, status_message, headers: hdrs, + http_version, }); loop { match response.chunk().await { diff --git a/crates/perry-ext-http/src/client_events.rs b/crates/perry-ext-http/src/client_events.rs index 88b89e2e88..8d27eb392f 100644 --- a/crates/perry-ext-http/src/client_events.rs +++ b/crates/perry-ext-http/src/client_events.rs @@ -256,6 +256,7 @@ pub(crate) unsafe fn handle_response_event( headers: Vec<(String, String)>, trailers: Vec<(String, String)>, body: Vec, + http_version: (u8, u8), ) { // #4909 — a destroyed request delivers nothing (Node tears the // exchange down); `completed` also suppresses any late timeout timer. @@ -291,6 +292,10 @@ pub(crate) unsafe fn handle_response_event( pipes: Vec::new(), socket_handle, request_handle, + http_version, + // Whole body already fully received by construction time (this is + // the synchronous single-event path). + complete: true, }); // Hand the IncomingMessage handle to the user's `(res) => { ... }` @@ -391,6 +396,96 @@ pub(crate) unsafe fn handle_response_event( fire_request_close_once(request_handle); } +/// Drain handler for `PendingHttpEvent::Upgrade` (#10468): build a +/// lightweight client `IncomingMessage` (statusCode/headers only — the body +/// is the upgraded protocol now, delivered over the adopted socket instead) +/// and fire `req.on('upgrade', (res, socket, head) => ...)` with +/// `(res, socket, head)`, Node's exact argument shape. `socket` is the +/// `net.Socket` id `client_upgrade::dispatch_upgrade_http_request` already +/// adopted via `perry_ext_net::adopt_upgraded_tcp_stream`; `head` is any +/// bytes the peer sent past the header block, as a `Buffer` (never a lossy +/// string — the write side of #10471 stays server-only, this is a fresh +/// client-side implementation). +/// +/// # Safety +/// +/// Same listener-liveness contract as [`fire_request_event_listeners`]. +pub(crate) unsafe fn handle_upgrade_event( + request_handle: Handle, + status: u16, + status_message: String, + headers: Vec<(String, String)>, + socket_handle: Handle, + head: Vec, +) { + let already_done = with_handle_mut::(request_handle, |req| { + let was = req.completed; + req.completed = true; + was + }) + .unwrap_or(true); + if already_done { + return; + } + client_abort::cleanup_request_signal(request_handle); + + // Main-thread companion of `adopt_upgraded_tcp_stream` (#4973) — must + // run before user code touches the socket. + if socket_handle != 0 { + perry_ext_net::ensure_adopted_socket_dispatch(); + } + + let incoming = register_handle(IncomingMessageHandle { + status_code: status, + status_message, + headers, + trailers: HashMap::new(), + body: Vec::new(), + listeners: HashMap::new(), + encoding: None, + decoder_pending: Vec::new(), + pipes: Vec::new(), + socket_handle, + request_handle, + http_version: (1, 1), + complete: true, + }); + + let upgrade_listeners = with_handle_mut::(request_handle, |req| { + take_request_event_listeners(req, "upgrade") + }) + .unwrap_or_default(); + + let res_arg = f64::from_bits(POINTER_TAG | (incoming as u64 & PTR_MASK)); + let socket_arg = if socket_handle == 0 { + f64::from_bits(TAG_UNDEFINED) + } else { + f64::from_bits(POINTER_TAG | (socket_handle as u64 & PTR_MASK)) + }; + // Node always hands the listener a Buffer here, even when the peer sent + // no bytes past the header block (`Buffer.isBuffer(head) === true` for a + // zero-length upgrade head) — never `undefined`. + let head_arg = { + let buf = perry_ffi::alloc_buffer(&head); + f64::from_bits(POINTER_TAG | (buf as u64 & PTR_MASK)) + }; + + let scope = perry_ffi::TransientRootScope::enter(); + let res_arg = scope.root_nanbox(res_arg); + let socket_arg = scope.root_nanbox(socket_arg); + let head_arg = scope.root_nanbox(head_arg); + let listeners = scope.root_addrs(&upgrade_listeners); + for cb in listeners { + if cb.get() != 0 { + let closure = JsClosure::from_raw(cb.get() as *const RawClosureHeader); + let _ = closure.call3(res_arg.get(), socket_arg.get(), head_arg.get()); + } + } + + finish_agent_request(request_handle, false); + fire_request_close_once(request_handle); +} + /// Drain handler for `PendingHttpEvent::ResponseHead` (streaming path): /// build the IncomingMessage handle with an empty body, remember it on the /// request, and fire the factory callback + `'response'` listeners. Body @@ -404,6 +499,7 @@ pub(crate) unsafe fn handle_response_head_event( status: u16, status_message: String, headers: Vec<(String, String)>, + http_version: (u8, u8), ) { // A destroyed request delivers nothing. let destroyed = @@ -428,6 +524,10 @@ pub(crate) unsafe fn handle_response_head_event( pipes: Vec::new(), socket_handle, request_handle, + http_version, + // The body streams in later (`ResponseChunk`/`ResponseEnd`); Node + // keeps `res.complete` false until the end edge. + complete: false, }); let (response_callback, response_listeners) = with_handle_mut::(request_handle, |request| { @@ -537,6 +637,11 @@ pub(crate) unsafe fn handle_response_end_event(request_handle: Handle) { return; } client_abort::cleanup_request_signal(request_handle); + // #10467 — the body has now been fully received; flip `res.complete` + // before the `'end'` listeners below observe it. + if let Some(im) = get_handle_mut::(incoming) { + im.complete = true; + } let (data_listeners, encoding, buffered, pipes) = get_handle_mut::(incoming) diff --git a/crates/perry-ext-http/src/client_surface.rs b/crates/perry-ext-http/src/client_surface.rs index 6058e89641..5dd4631863 100644 --- a/crates/perry-ext-http/src/client_surface.rs +++ b/crates/perry-ext-http/src/client_surface.rs @@ -224,6 +224,93 @@ pub extern "C" fn js_http_incoming_message_socket(handle: Handle) -> f64 { .unwrap_or_else(|| f64::from_bits(TAG_UNDEFINED)) } +/// `res.rawHeaders` (#10467) — see `build_raw_headers_array` for the +/// header-casing caveat on the pooled reqwest path. +#[no_mangle] +pub extern "C" fn js_http_response_raw_headers(handle: Handle) -> f64 { + let mut out = f64::from_bits(TAG_UNDEFINED); + with_handle_mut::(handle, |res| { + out = crate::response_headers::build_raw_headers_array(&res.headers); + }); + if out.to_bits() == TAG_UNDEFINED { + if let Some(server_out) = server_incoming_property(handle, "rawHeaders") { + return server_out; + } + } + out +} + +/// `res.httpVersion` — `"{major}.{minor}"` (#10467). The codegen native +/// table routes both client responses and server `IncomingMessage`s +/// through this entry (shared `class_filter`), so a registry miss here +/// falls back to the server accessor rather than defaulting blindly — +/// otherwise every server-side `req.httpVersion` would read back "1.1" +/// regardless of the real negotiated version. +#[no_mangle] +pub extern "C" fn js_http_response_http_version(handle: Handle) -> *mut StringHeader { + let mut out: Option = None; + with_handle_mut::(handle, |res| { + out = Some(format!("{}.{}", res.http_version.0, res.http_version.1)); + }); + if let Some(s) = out { + return alloc_string(&s).as_raw(); + } + if let Some(server_out) = server_incoming_property(handle, "httpVersion") { + let bits = server_out.to_bits(); + if bits >> 48 == 0x7FFF || bits >> 48 == 0x7FFD { + return (bits & PTR_MASK) as *mut StringHeader; + } + } + alloc_string("1.1").as_raw() +} + +/// `res.httpVersionMajor` (#10467). Same shared-`class_filter` fallback as +/// `js_http_response_http_version` — a server `req.httpVersionMajor` must +/// still resolve through the server accessor. +#[no_mangle] +pub extern "C" fn js_http_response_http_version_major(handle: Handle) -> f64 { + if let Some(v) = + with_handle_mut::(handle, |res| res.http_version.0 as f64) + { + return v; + } + server_incoming_property(handle, "httpVersionMajor").unwrap_or(1.0) +} + +/// `res.httpVersionMinor` (#10467). Same shared-`class_filter` fallback as +/// `js_http_response_http_version`. +#[no_mangle] +pub extern "C" fn js_http_response_http_version_minor(handle: Handle) -> f64 { + if let Some(v) = + with_handle_mut::(handle, |res| res.http_version.1 as f64) + { + return v; + } + server_incoming_property(handle, "httpVersionMinor").unwrap_or(1.0) +} + +/// `res.complete` (#10467) — `true` once the body has been fully received +/// (Node's aborted-download check). Same shared-`class_filter` fallback as +/// `js_http_response_http_version` — a server `req.complete` must still +/// resolve through the server accessor (`js_node_http_im_complete`, via +/// the dynamic dispatcher, which already returns a boxed JS boolean here). +#[no_mangle] +pub extern "C" fn js_http_response_complete(handle: Handle) -> f64 { + if let Some(v) = with_handle_mut::(handle, |res| { + if res.complete { + TAG_TRUE + } else { + TAG_FALSE + } + }) { + return f64::from_bits(v); + } + if let Some(server_out) = server_incoming_property(handle, "complete") { + return server_out; + } + f64::from_bits(TAG_UNDEFINED) +} + /// `res.req` — the ClientRequest paired with a client IncomingMessage. #[no_mangle] pub extern "C" fn js_http_incoming_message_req(handle: Handle) -> f64 { diff --git a/crates/perry-ext-http/src/client_upgrade.rs b/crates/perry-ext-http/src/client_upgrade.rs new file mode 100644 index 0000000000..f97975a492 --- /dev/null +++ b/crates/perry-ext-http/src/client_upgrade.rs @@ -0,0 +1,183 @@ +//! #10468 — client-side protocol upgrade (`Connection: Upgrade`). A `101 +//! Switching Protocols` response hands the caller the raw socket through +//! `req.on('upgrade', (res, socket, head) => ...)` instead of an ordinary +//! `'response'`. reqwest consumes the connection as a normal response body +//! and never exposes it, so an upgrade request speaks HTTP/1.1 over a raw +//! `TcpStream` instead — the same shape as the trailer-aware bypass in +//! `plain_client.rs` — and, on a `101`, adopts the stream into +//! `perry_ext_net` as a `net.Socket` (mirrors the server's +//! `server/raw_upgrade.rs`). +//! +//! Scope: plain `http://` only — TLS upgrade needs a different transport +//! and falls through to the normal path (pre-#10468 behavior: no upgrade), +//! same as when this module isn't triggered at all (no `Connection: +//! Upgrade`, or an Agent/`createConnection` override already claimed the +//! connection before `dispatch_request` runs). + +use std::collections::HashMap; + +use perry_ffi::Handle; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use crate::{push_event, PendingHttpEvent}; + +/// `true` if `headers` asks for a protocol upgrade — `Connection: Upgrade` +/// as one token of a comma list (RFC 7230 §6.1; Node/undici send it as a +/// bare `Upgrade` value in practice). +pub(crate) fn wants_upgrade(headers: &HashMap) -> bool { + headers.iter().any(|(name, value)| { + name.eq_ignore_ascii_case("connection") + && value + .split(',') + .any(|part| part.trim().eq_ignore_ascii_case("upgrade")) + }) +} + +/// Speak the request over a raw `TcpStream` when it wants a protocol +/// upgrade. `None` means "not applicable" (not an upgrade request, or +/// `https://` — fall through to the normal reqwest path); `Some(Ok(()))` +/// once the exchange has been fully handed off to a `PendingHttpEvent` +/// (`Upgrade` on `101`, `Response` otherwise); `Some(Err(_))` on a +/// transport failure. Mirrors `plain_client::dispatch_plain_http_request`'s +/// bypass contract. +pub(crate) async fn dispatch_upgrade_http_request( + request_handle: Handle, + method: &str, + url: &str, + headers: &HashMap, + body: &[u8], + timeout_ms: Option, +) -> Option> { + if !wants_upgrade(headers) { + return None; + } + let parsed = match reqwest::Url::parse(url) { + Ok(u) if u.scheme() == "http" => u, + // https:// upgrade isn't implemented — let the caller fall through + // rather than mishandle it here (matches pre-#10468 behavior for TLS). + _ => return None, + }; + let host = match parsed.host_str() { + Some(h) => h.to_string(), + None => return Some(Err("missing host".to_string())), + }; + let port = parsed.port_or_known_default().unwrap_or(80); + let mut path = parsed.path().to_string(); + if path.is_empty() { + path.push('/'); + } + if let Some(q) = parsed.query() { + path.push('?'); + path.push_str(q); + } + + let deadline = std::time::Duration::from_millis(timeout_ms.unwrap_or(30_000)); + let fut = async { + let mut stream = tokio::net::TcpStream::connect((host.as_str(), port)).await?; + let host_header = if parsed.port().is_some() { + format!("{}:{}", host, port) + } else { + host.clone() + }; + let mut req = format!("{} {} HTTP/1.1\r\nHost: {}\r\n", method, path, host_header); + let mut has_content_length = false; + for (k, v) in headers { + if k.eq_ignore_ascii_case("content-length") { + has_content_length = true; + } + req.push_str(k); + req.push_str(": "); + req.push_str(v); + req.push_str("\r\n"); + } + if !body.is_empty() && !has_content_length { + req.push_str(&format!("Content-Length: {}\r\n", body.len())); + } + req.push_str("\r\n"); + stream.write_all(req.as_bytes()).await?; + if !body.is_empty() { + stream.write_all(body).await?; + } + + // Read only up to the end of the header block — a `101` keeps the + // connection open for the upgraded protocol, so (unlike + // `plain_client`'s trailer-aware bypass) this must not read to EOF. + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + while !buf.windows(4).any(|w| w == b"\r\n\r\n") { + let n = stream.read(&mut chunk).await?; + if n == 0 { + break; + } + buf.extend_from_slice(&chunk[..n]); + } + Ok::<_, std::io::Error>((stream, buf)) + }; + + let (stream, buf) = match tokio::time::timeout(deadline, fut).await { + Ok(Ok(v)) => v, + Ok(Err(e)) => return Some(Err(e.to_string())), + Err(_) => return Some(Err("request timed out".to_string())), + }; + + let Some(header_end) = buf.windows(4).position(|w| w == b"\r\n\r\n") else { + return Some(Err( + "invalid HTTP response (no header terminator)".to_string() + )); + }; + let head_text = String::from_utf8_lossy(&buf[..header_end]); + let mut lines = head_text.split("\r\n"); + let status_line = lines.next().unwrap_or_default(); + let mut parts = status_line.splitn(3, ' '); + let http_version = parts + .next() + .and_then(|v| v.strip_prefix("HTTP/")) + .and_then(|v| v.split_once('.')) + .and_then(|(maj, min)| Some((maj.parse::().ok()?, min.parse::().ok()?))) + .unwrap_or((1, 1)); + let status: u16 = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0); + let status_message = parts.next().unwrap_or("").to_string(); + let mut hdrs = Vec::new(); + for line in lines { + if let Some((name, value)) = line.split_once(':') { + hdrs.push((name.trim().to_ascii_lowercase(), value.trim().to_string())); + } + } + let rest = buf[header_end + 4..].to_vec(); + + if status == 101 { + let socket_id = perry_ext_net::adopt_upgraded_tcp_stream(stream); + push_event(PendingHttpEvent::Upgrade { + request_handle, + status, + status_message, + headers: hdrs, + socket_handle: socket_id, + head: rest, + }); + return Some(Ok(())); + } + + // Server declined the upgrade — deliver an ordinary `'response'`. Read + // the remainder to EOF like the trailer-aware bypass (a non-101 reply + // to an Upgrade request has no further framing guarantee here). + let mut stream = stream; + let mut full = rest; + let mut chunk = [0u8; 16 * 1024]; + loop { + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => full.extend_from_slice(&chunk[..n]), + } + } + push_event(PendingHttpEvent::Response { + request_handle, + status, + status_message, + headers: hdrs, + trailers: Vec::new(), + body: full, + http_version, + }); + Some(Ok(())) +} diff --git a/crates/perry-ext-http/src/continue_client.rs b/crates/perry-ext-http/src/continue_client.rs index b0822a088e..baed123cdf 100644 --- a/crates/perry-ext-http/src/continue_client.rs +++ b/crates/perry-ext-http/src/continue_client.rs @@ -273,6 +273,7 @@ async fn run_exchange( status: parsed.status, status_message: parsed.status_message, headers: parsed.headers, + http_version: parsed.http_version, }); if !parsed.body.is_empty() { push_event(PendingHttpEvent::ResponseChunk { diff --git a/crates/perry-ext-http/src/lib.rs b/crates/perry-ext-http/src/lib.rs index 758a572f87..227f19825d 100644 --- a/crates/perry-ext-http/src/lib.rs +++ b/crates/perry-ext-http/src/lib.rs @@ -58,6 +58,8 @@ mod tls_client; // Raw-socket trailer-aware HTTP/1.1 client (`TE: trailers` bypass) + // response parser, extracted to keep `lib.rs` under the 2000-line lint cap. +mod client_connect_override; +mod client_upgrade; mod plain_client; use plain_client::{dispatch_plain_http_request, parse_http_response}; @@ -152,6 +154,7 @@ pub(crate) enum PendingHttpEvent { headers: Vec<(String, String)>, trailers: Vec<(String, String)>, body: Vec, + http_version: (u8, u8), }, /// Streaming delivery (reqwest path): the response head arrived — fire /// the `http.request` callback / `'response'` listeners now; body @@ -163,6 +166,7 @@ pub(crate) enum PendingHttpEvent { status: u16, status_message: String, headers: Vec<(String, String)>, + http_version: (u8, u8), }, /// One streamed body chunk following a `ResponseHead`. Carried as a /// refcounted `Bytes` (reqwest hands `chunk()` out this way) so the @@ -175,6 +179,15 @@ pub(crate) enum PendingHttpEvent { /// The streamed body finished — `'end'` on the message, `'close'` on /// the request. ResponseEnd { request_handle: Handle }, + /// #10468 — a `101` fires `'upgrade'` instead of `'response'` (`client_upgrade.rs`). + Upgrade { + request_handle: Handle, + status: u16, + status_message: String, + headers: Vec<(String, String)>, + socket_handle: Handle, + head: Vec, + }, Error { request_handle: Handle, error_message: String, @@ -494,6 +507,8 @@ pub struct ClientRequestHandle { /// options object. HTTPS TLS identity fields are lost if this is /// reconstructed from the URL at release time. agent_key: String, + /// `options.createConnection` (#10469, only when `agent_handle == 0`). + request_create_connection: i64, /// Agent pool bookkeeping. Exactly one of these is true after `end()` /// admits the request; terminal events clear `agent_active`, while a /// maxSockets waiter stays queued until the active request releases it. @@ -576,6 +591,10 @@ pub struct IncomingMessageHandle { /// ClientRequest that produced this response (`res.req`). Server-side /// IncomingMessages live in a separate registry and never populate this. pub request_handle: Handle, + /// `res.httpVersion*` (#10467); `(major, minor)`, default `(1, 1)`. + pub http_version: (u8, u8), + /// `res.complete` (#10467) — set once the body is fully received. + pub complete: bool, } unsafe impl Send for IncomingMessageHandle {} @@ -695,6 +714,7 @@ fn make_request_handle( callback: i64, agent_handle: Handle, agent_key: String, + request_create_connection: i64, ) -> Handle { let async_id = unsafe { js_async_hooks_provider_init(b"HTTPCLIENTREQUEST".as_ptr(), b"HTTPCLIENTREQUEST".len()) @@ -718,6 +738,7 @@ fn make_request_handle( close_emitted: false, agent_handle, agent_key, + request_create_connection, agent_active: false, agent_queued: false, reused_socket: false, @@ -795,6 +816,7 @@ fn pending_request_handle(event: &PendingHttpEvent) -> Handle { match event { PendingHttpEvent::Socket { request_handle } | PendingHttpEvent::SignalAbort { request_handle } + | PendingHttpEvent::Upgrade { request_handle, .. } | PendingHttpEvent::Response { request_handle, .. } | PendingHttpEvent::ResponseHead { request_handle, .. } | PendingHttpEvent::ResponseChunk { request_handle, .. } @@ -814,6 +836,7 @@ fn terminal_http_event(event: &PendingHttpEvent) -> bool { matches!( event, PendingHttpEvent::SignalAbort { .. } + | PendingHttpEvent::Upgrade { .. } | PendingHttpEvent::Response { .. } | PendingHttpEvent::ResponseEnd { .. } | PendingHttpEvent::Error { .. } @@ -926,166 +949,6 @@ fn tls_servername_from_host_header(value: &str) -> Option { } } -/// Serialize an HTTP/1.1 request (request line + headers + body) into the -/// bytes to write onto a socket. Forces `Connection: close` (the raw socket -/// path reads until EOF), drops any caller-supplied `Connection`/`Host` -/// header (we set `Host` from the URL), and adds `Content-Length` when a -/// body is present and the caller didn't. -fn serialize_http_request( - method: &str, - path: &str, - host_header: &str, - headers: &HashMap, - body: &[u8], -) -> Vec { - let mut req = format!("{} {} HTTP/1.1\r\nHost: {}\r\n", method, path, host_header); - let mut has_content_length = false; - for (k, v) in headers { - if k.eq_ignore_ascii_case("content-length") { - has_content_length = true; - } - if k.eq_ignore_ascii_case("connection") || k.eq_ignore_ascii_case("host") { - continue; - } - req.push_str(k); - req.push_str(": "); - req.push_str(v); - req.push_str("\r\n"); - } - req.push_str("Connection: close\r\n"); - if !body.is_empty() && !has_content_length { - req.push_str(&format!("Content-Length: {}\r\n", body.len())); - } - req.push_str("\r\n"); - let mut out = req.into_bytes(); - out.extend_from_slice(body); - out -} - -/// #2154 — run an HTTP exchange over a socket that the agent's -/// `createConnection` override produced (`socket_id`), instead of through -/// reqwest. Writes the serialized request, reads the response until the peer -/// closes (we force `Connection: close`), parses it with -/// [`parse_http_response`], and pushes the same `Response` / `Error` event -/// the reqwest path produces — so the IncomingMessage surface is identical. -/// -/// The socket I/O goes through perry-ffi's raw-net vtable (published by -/// perry-ext-net), so this crate needs no link edge to perry-ext-net. If no -/// net backend is linked the request errors out (the override couldn't have -/// produced a socket without `net`, so this is a defensive guard). -fn dispatch_request_over_socket( - request_handle: Handle, - method: String, - url: String, - headers: HashMap, - body: Vec, - timeout_ms: Option, - socket_id: i64, -) { - let parsed = match reqwest::Url::parse(&url) { - Ok(u) => u, - Err(e) => { - push_event(PendingHttpEvent::Error { - request_handle, - error_message: e.to_string(), - }); - return; - } - }; - let host = parsed.host_str().unwrap_or("localhost").to_string(); - let host_header = match parsed.port() { - Some(p) => format!("{}:{}", host, p), - None => host, - }; - let mut path = parsed.path().to_string(); - if path.is_empty() { - path.push('/'); - } - if let Some(q) = parsed.query() { - path.push('?'); - path.push_str(q); - } - let req_bytes = serialize_http_request(&method, &path, &host_header, &headers, &body); - let deadline = std::time::Duration::from_millis(timeout_ms.unwrap_or(30_000)); - - spawn_blocking(move || { - let try_h = tokio::runtime::Handle::try_current(); - std::hint::black_box(&try_h); - if try_h.is_err() { - push_event(PendingHttpEvent::Error { - request_handle, - error_message: "http client runtime unavailable".to_string(), - }); - return; - } - let handle = tokio::runtime::Handle::current(); - // #5779 follow-up: keep this fetch counted in-flight for its whole - // lifetime so the idle-kick recovers a lost worker-unpark. - let inflight_guard = ClientInflightGuard::new(request_handle); - let jh = handle.spawn(async move { - let _inflight = inflight_guard; - let vtable = match perry_ffi::raw_net() { - Some(v) => v, - None => { - push_event(PendingHttpEvent::Error { - request_handle, - error_message: "agent.createConnection requires node:net (not linked)" - .to_string(), - }); - return; - } - }; - // Attach is idempotent — the request path also attaches on the - // main thread before this task runs, to close any data race. - (vtable.attach)(socket_id); - if (vtable.write)(socket_id, req_bytes.as_ptr(), req_bytes.len()) == 0 { - push_event(PendingHttpEvent::Error { - request_handle, - error_message: "failed to write request to agent socket".to_string(), - }); - return; - } - - let mut raw = Vec::new(); - let mut chunk = [0u8; 16 * 1024]; - let start = tokio::time::Instant::now(); - loop { - let n = (vtable.poll_read)(socket_id, chunk.as_mut_ptr(), chunk.len()); - if n > 0 { - raw.extend_from_slice(&chunk[..n as usize]); - } else if n == 0 { - break; // clean EOF — peer closed after the response - } else { - if start.elapsed() >= deadline { - (vtable.close)(socket_id); - push_event(PendingHttpEvent::Timeout { request_handle }); - return; - } - tokio::time::sleep(std::time::Duration::from_millis(1)).await; - } - } - (vtable.close)(socket_id); - - match parse_http_response(&raw) { - Ok(parsed) => push_event(PendingHttpEvent::Response { - request_handle, - status: parsed.status, - status_message: parsed.status_message, - headers: parsed.headers, - trailers: parsed.trailers, - body: parsed.body, - }), - Err(error_message) => push_event(PendingHttpEvent::Error { - request_handle, - error_message, - }), - } - }); - std::hint::black_box(&jh); - std::mem::forget(jh); - }); -} - /// #2154 — invoke a user `createSocket(req, options, cb)` override on the /// request path (Node's `Agent.prototype.addRequest` semantics). Builds the /// three arguments Node passes: @@ -1136,7 +999,12 @@ unsafe fn invoke_create_socket( let cb = (cb_val.get().to_bits() & PTR_MASK) as *mut perry_ffi::ClosureHeader; perry_ffi::set_closure_capture_f64(cb, 0, request_handle as f64); let req_val = f64::from_bits(POINTER_TAG | (request_handle as u64 & PTR_MASK)); - let options = scope.root_nanbox(agent::build_connect_options(agent_handle, host, port, path)); + let options = scope.root_nanbox(agent::build_connect_options( + Some(agent_handle), + host, + port, + path, + )); let closure = JsClosure::from_raw(cs.get() as *const RawClosureHeader); closure.call3(req_val, options.get(), cb_val.get()); @@ -1210,7 +1078,7 @@ unsafe extern "C" fn http_create_socket_cb( if let Some(vt) = perry_ffi::raw_net() { (vt.attach)(socket_id); } - dispatch_request_over_socket( + client_connect_override::dispatch_request_over_socket( request_handle, method, url, @@ -1262,6 +1130,7 @@ unsafe fn request_common(arg_f64: f64, callback: i64, default_protocol: &str) -> agent_handle }; let agent_key = agent::request_key_from_options(agent_handle, arg_f64, &url); + let request_create_connection = agent::request_create_connection_from_options(arg_f64); // #10469 let handle = make_request_handle( method, url, @@ -1270,6 +1139,7 @@ unsafe fn request_common(arg_f64: f64, callback: i64, default_protocol: &str) -> callback, agent_handle, agent_key, + request_create_connection, ); client_abort::attach_request_signal(handle, arg_f64); attach_tls_options(handle, arg_f64); // #4906 @@ -1327,6 +1197,7 @@ unsafe fn get_common(arg_f64: f64, callback: i64, default_protocol: &str) -> Han agent_handle }; let agent_key = agent::request_key_from_options(agent_handle, arg_f64, &url); + let request_create_connection = agent::request_create_connection_from_options(arg_f64); // #10469 let handle = make_request_handle( "GET".to_string(), url, @@ -1335,6 +1206,7 @@ unsafe fn get_common(arg_f64: f64, callback: i64, default_protocol: &str) -> Han callback, agent_handle, agent_key, + request_create_connection, ); client_abort::attach_request_signal(handle, arg_f64); attach_tls_options(handle, arg_f64); // #4906 @@ -1388,6 +1260,7 @@ unsafe fn request_overload(args_array: i64, default_protocol: &str, force_get: b agent_handle }; let agent_key = agent::request_key_from_options(agent_handle, parsed.opts, &url); + let request_create_connection = agent::request_create_connection_from_options(parsed.opts); // #10469 let handle = make_request_handle( method, url, @@ -1396,6 +1269,7 @@ unsafe fn request_overload(args_array: i64, default_protocol: &str, force_get: b parsed.callback, agent_handle, agent_key, + request_create_connection, ); client_abort::attach_request_signal(handle, parsed.opts); attach_tls_options(handle, parsed.opts); // #4906 — TLS options ride on the options bag @@ -1706,7 +1580,7 @@ unsafe fn dispatch_request_snapshot(handle: Handle, snapshot: RequestSnapshot) { if let Some(vt) = perry_ffi::raw_net() { (vt.attach)(socket_id); } - dispatch_request_over_socket( + client_connect_override::dispatch_request_over_socket( handle, method, url, headers, body, timeout_ms, socket_id, ); return; @@ -1714,6 +1588,16 @@ unsafe fn dispatch_request_snapshot(handle: Handle, snapshot: RequestSnapshot) { } } + // #10469 — request-level `createConnection` (no explicit Agent). + if agent_handle == 0 { + if let Some(socket_id) = client_connect_override::dispatch_for_handle(handle, &url) { + client_connect_override::dispatch_request_over_socket( + handle, method, url, headers, body, timeout_ms, socket_id, + ); + return; + } + } + dispatch_request( handle, method, diff --git a/crates/perry-ext-http/src/pending_dispatch.rs b/crates/perry-ext-http/src/pending_dispatch.rs index c2a1bb24cd..27b50380ca 100644 --- a/crates/perry-ext-http/src/pending_dispatch.rs +++ b/crates/perry-ext-http/src/pending_dispatch.rs @@ -48,6 +48,7 @@ pub unsafe extern "C" fn js_http_process_pending() -> i32 { headers, trailers, body, + http_version, } => client_events::handle_response_event( request_handle, status, @@ -55,17 +56,35 @@ pub unsafe extern "C" fn js_http_process_pending() -> i32 { headers, trailers, body, + http_version, ), PendingHttpEvent::ResponseHead { request_handle, status, status_message, headers, + http_version, } => client_events::handle_response_head_event( request_handle, status, status_message, headers, + http_version, + ), + PendingHttpEvent::Upgrade { + request_handle, + status, + status_message, + headers, + socket_handle, + head, + } => client_events::handle_upgrade_event( + request_handle, + status, + status_message, + headers, + socket_handle, + head, ), PendingHttpEvent::ResponseChunk { request_handle, diff --git a/crates/perry-ext-http/src/plain_client.rs b/crates/perry-ext-http/src/plain_client.rs index 2f0d3cd304..d1b2939ddf 100644 --- a/crates/perry-ext-http/src/plain_client.rs +++ b/crates/perry-ext-http/src/plain_client.rs @@ -112,6 +112,7 @@ pub(crate) async fn dispatch_plain_http_request( headers: parsed.headers, trailers: parsed.trailers, body: parsed.body, + http_version: parsed.http_version, }); Some(Ok(())) } @@ -127,6 +128,10 @@ pub(crate) struct ParsedHttpResponse { pub(crate) headers: Vec<(String, String)>, pub(crate) trailers: Vec<(String, String)>, pub(crate) body: Vec, + /// `(major, minor)` parsed from the status line (`HTTP/1.1 200 OK`). + /// Falls back to `(1, 1)` on anything that doesn't parse as + /// `HTTP/.` (#10467). + pub(crate) http_version: (u8, u8), } /// Parse a raw HTTP/1.1 response (the bytes read off a socket) into status / @@ -144,7 +149,12 @@ pub(crate) fn parse_http_response(raw: &[u8]) -> Result().ok()?, min.parse::().ok()?))) + .unwrap_or((1, 1)); let status = status_parts .next() .and_then(|s| s.parse::().ok()) @@ -215,5 +225,6 @@ pub(crate) fn parse_http_response(raw: &[u8]) -> Result bool { ) } +/// Build `res.rawHeaders` (#10467) — the flattened `[name, value, name, +/// value, ...]` array in wire arrival order, duplicates preserved (unlike +/// the combined `headers` view above, which merges/collapses per +/// `matchKnownFields`). +/// +/// Caveat: header name casing here is whatever the transport captured. The +/// pooled reqwest path normalizes names to lower case before Perry ever +/// sees them (`http::HeaderName` only stores lower case), so this does not +/// reproduce Node's original wire casing on that path — only the raw-socket +/// paths (`plain_client`/`agent.createConnection`) could preserve it, and +/// today they lower-case on parse too. Tracked as a known gap, not silently +/// papered over. +pub(crate) fn build_raw_headers_array(raw: &[(String, String)]) -> f64 { + let arr = unsafe { perry_ffi::js_array_alloc((raw.len() * 2) as u32) }; + if arr.is_null() { + return f64::from_bits(TAG_UNDEFINED); + } + // #10668-followup: `arr` is a raw heap pointer. `alloc_string` below can + // allocate (and therefore collect), which can move the array this + // pointer refers to before the next `js_array_push` reads it back. Root + // it through a `TransientRootScope` and re-derive the pointer via + // `.get()` after every allocating call instead of reusing the pre-call + // copy (see `docs/src/internals/gc-rooting-invariant.md`). Each pointer + // is materialized on its own line and consumed immediately by the + // `js_array_push` call on the very next line -- keep it that shape + // (not folded into a multi-line call) so no raw pointer is ever bound + // across the loop's next `alloc_string`. + let scope = TransientRootScope::enter(); + let mut arr = scope.root_nanbox(f64::from_bits(JsValue::from_object_ptr(arr).bits())); + for (name, value) in raw { + let name_s = alloc_string(name); + let arr_ptr = JsValue::from_bits(arr.get().to_bits()).as_pointer(); + let name_value = JsValue::from_string_ptr(name_s.as_raw()); + let pushed = unsafe { js_array_push(arr_ptr, name_value) }; + arr = scope.root_nanbox(f64::from_bits(JsValue::from_object_ptr(pushed).bits())); + + let value_s = alloc_string(value); + let arr_ptr = JsValue::from_bits(arr.get().to_bits()).as_pointer(); + let value_value = JsValue::from_string_ptr(value_s.as_raw()); + let pushed = unsafe { js_array_push(arr_ptr, value_value) }; + arr = scope.root_nanbox(f64::from_bits(JsValue::from_object_ptr(pushed).bits())); + } + arr.get() +} + +/// Build the `set-cookie` array for [`build_response_headers_object`] -- +/// always a string array, even for a single cookie (Node's +/// `matchKnownFields` never collapses `set-cookie`). Same GC-rooting shape +/// as [`build_raw_headers_array`]: `alloc_string` can collect and move +/// `arr` before `js_array_push` reads it back, so root and re-derive the +/// pointer at each use instead of reusing the pre-call copy. Kept as its +/// own top-level function (rather than nested inside the caller's +/// `if key == "set-cookie"` arm) so these lines stay short enough that +/// rustfmt doesn't wrap a `let` binding across lines and defeat the +/// ratchet scanner's line-oriented binding detection (#10668-followup). +fn build_set_cookie_array(set_cookie: &[String]) -> f64 { + let arr = unsafe { js_array_alloc(set_cookie.len() as u32) }; + let scope = TransientRootScope::enter(); + let mut arr = scope.root_nanbox(f64::from_bits(JsValue::from_object_ptr(arr).bits())); + for cookie in set_cookie { + let cookie_s = alloc_string(cookie); + let arr_ptr = JsValue::from_bits(arr.get().to_bits()).as_pointer(); + let cookie_value = JsValue::from_string_ptr(cookie_s.as_raw()); + let pushed = unsafe { js_array_push(arr_ptr, cookie_value) }; + arr = scope.root_nanbox(f64::from_bits(JsValue::from_object_ptr(pushed).bits())); + } + arr.get() +} + /// Build the combined `IncomingMessage.headers` object from the raw /// `(name, value)` pairs, applying Node's `matchKnownFields` rules /// (#5079): @@ -95,13 +166,7 @@ pub(crate) fn build_response_headers_object(raw: &[(String, String)]) -> f64 { if !obj.is_null() { for (i, key) in order.iter().enumerate() { let v = if key == "set-cookie" { - let mut arr = unsafe { js_array_alloc(set_cookie.len() as u32) }; - for cookie in &set_cookie { - arr = unsafe { - js_array_push(arr, JsValue::from_string_ptr(alloc_string(cookie).as_raw())) - }; - } - JsValue::from_object_ptr(arr) + JsValue::from_bits(build_set_cookie_array(&set_cookie).to_bits()) } else if let Some(val) = combined.get(key) { let s = alloc_string(val); JsValue::from_string_ptr(s.as_raw()) diff --git a/crates/perry-ext-http/src/tests.rs b/crates/perry-ext-http/src/tests.rs index dc7625ebff..00b4691138 100644 --- a/crates/perry-ext-http/src/tests.rs +++ b/crates/perry-ext-http/src/tests.rs @@ -99,6 +99,7 @@ fn gc_mutable_scanner_rewrites_request_response_listener_roots() { close_emitted: false, agent_handle: 0, agent_key: "localhost::".to_string(), + request_create_connection: 0, agent_active: false, agent_queued: false, reused_socket: false, @@ -126,6 +127,8 @@ fn gc_mutable_scanner_rewrites_request_response_listener_roots() { pipes: Vec::new(), socket_handle: 0, request_handle, + http_version: (1, 1), + complete: true, }); let _ = perry_runtime::gc::gc_collect_minor(); @@ -187,6 +190,7 @@ fn drain_streamed_body(chunks: &[&[u8]]) -> Vec { close_emitted: false, agent_handle: 0, agent_key: "localhost::".to_string(), + request_create_connection: 0, agent_active: false, agent_queued: false, reused_socket: false, @@ -208,6 +212,7 @@ fn drain_streamed_body(chunks: &[&[u8]]) -> Vec { 200, "OK".to_string(), Vec::new(), + (1, 1), ); // Each production chunk is a refcounted `Bytes` (reqwest's // `response.chunk()` shape) — build the input the same way so the @@ -359,6 +364,7 @@ fn dispatch_request_stays_visible_to_exit_gate_until_response_queued() { close_emitted: false, agent_handle: 0, agent_key: "localhost::".to_string(), + request_create_connection: 0, agent_active: false, agent_queued: false, reused_socket: false, diff --git a/crates/perry-stdlib/src/common/dispatch_http.rs b/crates/perry-stdlib/src/common/dispatch_http.rs index aae66413fd..df760d8a29 100644 --- a/crates/perry-stdlib/src/common/dispatch_http.rs +++ b/crates/perry-stdlib/src/common/dispatch_http.rs @@ -224,6 +224,12 @@ pub(super) unsafe fn dispatch_client_incoming_property( | "socket" | "connection" | "req" + // #10467 — rawHeaders / httpVersion* / complete. + | "rawHeaders" + | "httpVersion" + | "httpVersionMajor" + | "httpVersionMinor" + | "complete" ) { return None; } @@ -241,6 +247,11 @@ pub(super) unsafe fn dispatch_client_incoming_property( fn js_http_response_trailers(handle: i64) -> f64; fn js_http_incoming_message_socket(handle: i64) -> f64; fn js_http_incoming_message_req(handle: i64) -> f64; + fn js_http_response_raw_headers(handle: i64) -> f64; + fn js_http_response_http_version(handle: i64) -> *mut perry_runtime::StringHeader; + fn js_http_response_http_version_major(handle: i64) -> f64; + fn js_http_response_http_version_minor(handle: i64) -> f64; + fn js_http_response_complete(handle: i64) -> f64; } if unsafe { js_http_is_incoming_message(handle) } == 0 { @@ -267,6 +278,18 @@ pub(super) unsafe fn dispatch_client_incoming_property( "trailers" => unsafe { js_http_response_trailers(handle) }, "socket" | "connection" => unsafe { js_http_incoming_message_socket(handle) }, "req" => unsafe { js_http_incoming_message_req(handle) }, + "rawHeaders" => unsafe { js_http_response_raw_headers(handle) }, + "httpVersion" => { + let ptr = unsafe { js_http_response_http_version(handle) }; + if ptr.is_null() { + f64::from_bits(0x7FFC_0000_0000_0001) + } else { + f64::from_bits(JSValue::string_ptr(ptr).bits()) + } + } + "httpVersionMajor" => unsafe { js_http_response_http_version_major(handle) }, + "httpVersionMinor" => unsafe { js_http_response_http_version_minor(handle) }, + "complete" => unsafe { js_http_response_complete(handle) }, _ => f64::from_bits(0x7FFC_0000_0000_0001), }; Some(value) diff --git a/scripts/unrooted_local_shape_baseline.json b/scripts/unrooted_local_shape_baseline.json index 40f260534c..57e2872e30 100644 --- a/scripts/unrooted_local_shape_baseline.json +++ b/scripts/unrooted_local_shape_baseline.json @@ -13,7 +13,6 @@ "crates/perry-ext-fetch/src/tests.rs": 14, "crates/perry-ext-http/src/agent.rs": 3, "crates/perry-ext-http/src/client_request_surface.rs": 2, - "crates/perry-ext-http/src/response_headers.rs": 1, "crates/perry-ext-http/src/server/handle_dispatch.rs": 2, "crates/perry-ext-http/src/server/request.rs": 7, "crates/perry-ext-http/src/server/response.rs": 1, @@ -55,7 +54,7 @@ "crates/perry-stdlib/src/pg/types.rs": 14, "crates/perry-stdlib/src/querystring.rs": 2, "crates/perry-stdlib/src/ratelimit.rs": 4, - "crates/perry-stdlib/src/readline/mod.rs": 5, + "crates/perry-stdlib/src/readline/mod.rs": 4, "crates/perry-stdlib/src/sqlite/backup.rs": 7, "crates/perry-stdlib/src/sqlite/better.rs": 18, "crates/perry-stdlib/src/sqlite/bind.rs": 4, @@ -70,7 +69,7 @@ "crates/perry-stdlib/src/streams/transform.rs": 8, "crates/perry-stdlib/src/streams/writable.rs": 2, "crates/perry-stdlib/src/string_decoder.rs": 4, - "crates/perry-stdlib/src/tls.rs": 4, + "crates/perry-stdlib/src/tls.rs": 3, "crates/perry-stdlib/src/webcrypto/aes.rs": 2, "crates/perry-stdlib/src/webcrypto/encapsulation.rs": 8, "crates/perry-stdlib/src/webcrypto/jwk.rs": 2, @@ -83,5 +82,5 @@ "crates/perry-stdlib/src/zlib.rs": 2 }, "schema_version": 2, - "total": 561 + "total": 558 }