From 263b47ab2b880e8c29834c89f899907c069f5053 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 11 Jun 2026 06:48:57 +0200 Subject: [PATCH] fix(http): client write/end callbacks + backpressure, real client timeouts, dynamic listener registration (#4909) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sub-tickets (1) and (2) of the #4909 silent-hang triage, mirroring the #4954 server-side fix onto the perry-ext-http client path: - req.write(chunk[, enc][, cb]) queues its callback and returns a real NaN-boxed backpressure boolean (false past the 16 KiB high-water mark), so `while (req.write(buf, cb))` producer loops terminate. Buffer chunks are read through a buffer-aware reader instead of being misparsed as StringHeaders (same layout confusion as #1124). - req.end([chunk][, enc][, cb]) handles the end(cb) form and fires the queued write callbacks -> 'finish' -> end callback in Node's flush order via a new PendingHttpEvent::Flushed drained by the client pump. - req.setTimeout(ms[, cb]) arms a real timer (tokio sleep -> Timeout event) instead of only storing the delay, so 'timeout' fires even for a request that was never dispatched or whose server never responds; options.timeout arms the same timer at request creation. 'timeout' is emitted at most once and never after completion; setTimeout(0) clears the timer per Node. - req.destroy() on an in-flight request emits the coded ECONNRESET "socket hang up" then 'close' (once-only via close_emitted); the response/error drains suppress events for completed/destroyed requests; '.on("response")' listeners now fire alongside the factory callback. - Dynamic dispatch (untyped receivers): on/once/addListener/ removeListener/removeAllListeners now register/remove client-request listeners (the stdlib dispatch_http name-gate hid "on" so listeners were silently dropped), and `constructor` returns a `{ name }` object so out.constructor.name discriminates ClientRequest/ServerResponse/ IncomingMessage. - Server side: js_node_http_im_resume no longer consumes the one-shot 'end'/'data' emit flags when no listener is registered yet, fixing the canonical `req.resume(); req.on('end', cb)` drain pattern that hung every response; js_node_http_res_write_with_cb returns the same backpressure boolean as the static _full path; js_node_http_res_end_with_cb fires write cbs -> 'finish' -> end cb -> 'close' in Node order; and closure_arg validates with js_value_is_closure so a Buffer chunk in `res.end(buf, cb)` is no longer invoked as the callback (TypeError). Corpus bucket (#4909, 108 tests, baseline vs fixed on the same comparator): +8 fail->pass (outgoing-finish, client-timeout-event, agent-uninitialized x2, early-hints, client-race x2 and timeout-client-warning — the latter three byte-differ only by Node's DEP0169/pid noise the radar normalizes), 0 regressions; most remaining runtime-fails now exit with a real assertion message instead of the silent hang this issue tracks. Found while triaging (pre-existing, unchanged): test-http-response-setheaders SIGSEGVs nondeterministically on baseline and fixed builds alike. --- crates/perry-codegen/src/ext_registry.rs | 3 + .../lower_call/native_table/http_client.rs | 27 +- .../src/runtime_decls/stdlib_ffi.rs | 13 + .../src/handle_dispatch.rs | 27 ++ crates/perry-ext-http-server/src/request.rs | 18 +- crates/perry-ext-http-server/src/response.rs | 20 +- crates/perry-ext-http/src/client_events.rs | 259 +++++++++++++++--- crates/perry-ext-http/src/client_outgoing.rs | 195 +++++++++++++ .../src/client_request_surface.rs | 132 ++++++++- crates/perry-ext-http/src/lib.rs | 180 ++++++------ crates/perry-ext-http/src/tests.rs | 5 + crates/perry-stdlib/src/common/dispatch.rs | 2 + .../perry-stdlib/src/common/dispatch_http.rs | 18 ++ .../issue_4909_client_write_end_timeouts.rs | 155 +++++++++++ 14 files changed, 889 insertions(+), 165 deletions(-) create mode 100644 crates/perry-ext-http/src/client_outgoing.rs create mode 100644 crates/perry/tests/issue_4909_client_write_end_timeouts.rs diff --git a/crates/perry-codegen/src/ext_registry.rs b/crates/perry-codegen/src/ext_registry.rs index 1c48f73c53..c0fc6c7ab0 100644 --- a/crates/perry-codegen/src/ext_registry.rs +++ b/crates/perry-codegen/src/ext_registry.rs @@ -205,8 +205,11 @@ const FFI_REGISTRY: &[(&str, OwnerKind)] = &[ ("js_http_on", OwnerKind::WellKnown("http")), ("js_http_set_header", OwnerKind::WellKnown("http")), ("js_http_set_timeout", OwnerKind::WellKnown("http")), + ("js_http_set_timeout_full", OwnerKind::WellKnown("http")), ("js_http_client_request_end", OwnerKind::WellKnown("http")), ("js_http_client_request_write", OwnerKind::WellKnown("http")), + ("js_http_client_request_end_full", OwnerKind::WellKnown("http")), + ("js_http_client_request_write_full", OwnerKind::WellKnown("http")), ("js_http_client_request_listener_count", OwnerKind::WellKnown("http")), ("js_http_client_request_get_header", OwnerKind::WellKnown("http")), ("js_http_client_request_has_header", OwnerKind::WellKnown("http")), diff --git a/crates/perry-codegen/src/lower_call/native_table/http_client.rs b/crates/perry-codegen/src/lower_call/native_table/http_client.rs index 8746887a54..fd8841e9ad 100644 --- a/crates/perry-codegen/src/lower_call/native_table/http_client.rs +++ b/crates/perry-codegen/src/lower_call/native_table/http_client.rs @@ -141,8 +141,12 @@ pub(super) const HTTP_CLIENT_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "end", class_filter: Some("ClientRequest"), - runtime: "js_http_client_request_end", - args: &[NA_F64], + // #4909: route to the callback-aware end so `req.end(chunk, cb)` / + // `req.end(cb)` fire their callback (and the queued write callbacks + // + `'finish'`) in Node's flush order. arg2/arg3 carry the + // `(encoding?, callback?)` tail. + runtime: "js_http_client_request_end_full", + args: &[NA_F64, NA_JSV, NA_JSV], ret: NR_PTR, }, NativeModSig { @@ -150,9 +154,15 @@ pub(super) const HTTP_CLIENT_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "write", class_filter: Some("ClientRequest"), - runtime: "js_http_client_request_write", - args: &[NA_F64], - ret: NR_PTR, + // #4909: pass the trailing `(encoding?, callback?)` args as raw + // JSValues so the runtime can queue the write callback and return a + // real boolean (NR_F64 → NaN-boxed bool) for backpressure. The + // previous `js_http_client_request_write` (NA_F64 only / NR_PTR) + // dropped the callback and returned the always-truthy handle, so + // `while (req.write(buf))` producer loops never terminated. + runtime: "js_http_client_request_write_full", + args: &[NA_F64, NA_JSV, NA_JSV], + ret: NR_F64, }, NativeModSig { module: "http", @@ -168,8 +178,11 @@ pub(super) const HTTP_CLIENT_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "setTimeout", class_filter: Some("ClientRequest"), - runtime: "js_http_set_timeout", - args: &[NA_F64], + // #4909: the callback arg registers as a `'timeout'` listener and a + // real timer is armed (previously the delay was only stored, so + // `req.setTimeout(n, cb)` on a never-responding server hung forever). + runtime: "js_http_set_timeout_full", + args: &[NA_F64, NA_JSV], ret: NR_PTR, }, NativeModSig { diff --git a/crates/perry-codegen/src/runtime_decls/stdlib_ffi.rs b/crates/perry-codegen/src/runtime_decls/stdlib_ffi.rs index 6f0e94f4e6..5f2272a232 100644 --- a/crates/perry-codegen/src/runtime_decls/stdlib_ffi.rs +++ b/crates/perry-codegen/src/runtime_decls/stdlib_ffi.rs @@ -55,6 +55,19 @@ pub fn declare_stdlib_ffi(module: &mut LlModule) { // ========== HTTP server ========== module.declare_function("js_http_client_request_end", I64, &[I64, DOUBLE]); module.declare_function("js_http_client_request_write", I64, &[I64, DOUBLE]); + // #4909 — callback-aware client write/end/setTimeout (the `(encoding?, + // callback?)` tail rides as raw NaN-boxed JSValues). + module.declare_function( + "js_http_client_request_end_full", + I64, + &[I64, DOUBLE, I64, I64], + ); + module.declare_function( + "js_http_client_request_write_full", + DOUBLE, + &[I64, DOUBLE, I64, I64], + ); + module.declare_function("js_http_set_timeout_full", I64, &[I64, DOUBLE, I64]); module.declare_function("js_http_client_request_method", I64, &[I64]); module.declare_function("js_http_client_request_protocol", I64, &[I64]); module.declare_function("js_http_client_request_host", I64, &[I64]); diff --git a/crates/perry-ext-http-server/src/handle_dispatch.rs b/crates/perry-ext-http-server/src/handle_dispatch.rs index e22759d657..ecc80c2555 100644 --- a/crates/perry-ext-http-server/src/handle_dispatch.rs +++ b/crates/perry-ext-http-server/src/handle_dispatch.rs @@ -777,6 +777,7 @@ pub unsafe extern "C" fn js_ext_http_incoming_message_dispatch_property( "remoteAddress" => string_ptr_value(js_node_http_im_remote_address(handle)), "remotePort" => js_node_http_im_remote_port(handle), "rawBody" => js_node_http_im_raw_body(handle), + "constructor" => constructor_object("IncomingMessage"), _ => undef, } } @@ -819,10 +820,30 @@ pub unsafe extern "C" fn js_ext_http_server_response_dispatch_property( "strictContentLength" => bool_value(js_node_http_res_strict_content_length(handle) != 0), "req" => handle_value_or_undefined(js_node_http_res_req_handle(handle)), "socket" | "connection" => response_socket_value(handle), + // #4909 — `out.constructor.name` discrimination (corpus + // outgoing-message tests branch on it). + "constructor" => constructor_object("ServerResponse"), _ => undef, } } +/// `{ name: }` — stands in for `.constructor` so +/// `out.constructor.name` reads "ServerResponse"/"IncomingMessage" the way +/// the corpus outgoing-message tests expect (#4909). +fn constructor_object(name: &str) -> f64 { + let (packed, shape_id) = perry_ffi::build_object_shape(&["name"]); + let obj = + unsafe { js_object_alloc_with_shape(shape_id, 1, packed.as_ptr(), packed.len() as u32) }; + if obj.is_null() { + return f64::from_bits(TAG_UNDEFINED); + } + let value = JsValue::from_string_ptr(alloc_string(name).as_raw()); + unsafe { + perry_ffi::js_object_set_field(obj, 0, value); + } + f64::from_bits(JsValue::from_object_ptr(obj as *mut u8).bits()) +} + /// Dispatch a property write on a registered server-side `ServerResponse`. /// /// Returns 1 when the property was claimed. @@ -1097,6 +1118,12 @@ fn closure_arg(value: Option) -> i64 { if tag != 0x7FFD { return 0; } + // #4909 — a Buffer chunk is POINTER_TAG too; `end(buf, cb)` used to + // treat the buffer as the `end(cb)` callback form, drop the chunk, and + // then call the buffer ("TypeError: value is not a function"). + if unsafe { crate::types::js_value_is_closure(bits as i64) } == 0 { + return 0; + } (bits & PTR_MASK) as i64 } diff --git a/crates/perry-ext-http-server/src/request.rs b/crates/perry-ext-http-server/src/request.rs index f11c9eade0..34e4f26363 100644 --- a/crates/perry-ext-http-server/src/request.rs +++ b/crates/perry-ext-http-server/src/request.rs @@ -391,20 +391,22 @@ pub extern "C" fn js_node_http_im_resume(handle: i64) { let encoding; if let Some(im) = get_handle_mut::(handle) { im.paused = false; - if !im.data_emitted && !im.body_bytes.is_empty() { - should_emit_data = true; - } - if !im.end_emitted { - should_emit_end = true; - } body_bytes = im.body_bytes.clone(); data_listeners = im.listeners.get("data").cloned().unwrap_or_default(); end_listeners = im.listeners.get("end").cloned().unwrap_or_default(); encoding = im.encoding.clone(); - if should_emit_data { + // #4909 — only consume the one-shot emit flags when a listener is + // actually present. `req.resume(); req.on('end', cb)` (the canonical + // body-drain pattern) registers the listener one statement AFTER + // resume; marking `end_emitted` here meant `im_on`'s sync-emit arm + // saw the event as already delivered and the server hung without + // ever responding. + if !im.data_emitted && !im.body_bytes.is_empty() && !data_listeners.is_empty() { + should_emit_data = true; im.data_emitted = true; } - if should_emit_end { + if !im.end_emitted && !end_listeners.is_empty() { + should_emit_end = true; im.end_emitted = true; im.complete = true; } diff --git a/crates/perry-ext-http-server/src/response.rs b/crates/perry-ext-http-server/src/response.rs index 8a5e9ab579..45df296df5 100644 --- a/crates/perry-ext-http-server/src/response.rs +++ b/crates/perry-ext-http-server/src/response.rs @@ -1255,6 +1255,10 @@ pub extern "C" fn js_node_http_res_detach_socket(handle: i64, _socket: f64) { #[no_mangle] pub extern "C" fn js_node_http_res_write_with_cb(handle: i64, chunk: f64, callback: i64) -> i32 { let bytes = jsvalue_to_body_bytes(chunk); + // #4909 — real backpressure boolean (mirrors `js_node_http_res_write_full` + // on the static path): `false` past the 16 KiB high-water mark, so dynamic + // `while (res.write(buf, cb))` producer loops terminate. + let mut below_hwm = true; if let Some(sr) = get_handle_mut::(handle) { if !sr.writable_ended { sr.headers_sent = true; @@ -1264,9 +1268,14 @@ pub extern "C" fn js_node_http_res_write_with_cb(handle: i64, chunk: f64, callba if callback != 0 { sr.pending_write_callbacks.push(callback); } + below_hwm = sr.buffered_body.len() <= DEFAULT_HIGH_WATER_MARK; } } - 1 + if below_hwm { + 1 + } else { + 0 + } } /// `res.end([chunk][, callback])` — callback-aware variant. Standalone @@ -1282,16 +1291,23 @@ pub unsafe extern "C" fn js_node_http_res_end_with_cb(handle: i64, chunk: f64, c standalone_end(handle, chunk, callback); return; } - js_node_http_res_end(handle, chunk); + // #4909 — Node's flush ordering, matching `js_node_http_res_end_full`: + // queued write callbacks → `'finish'` → end callback → `'close'`. The + // previous code fired `'finish'`/`'close'` (via `js_node_http_res_end`) + // before any callback ran. + let listeners = finalize_buffered_end(handle, chunk); let write_cbs = get_handle_mut::(handle) .map(|sr| std::mem::take(&mut sr.pending_write_callbacks)) .unwrap_or_default(); for cb in write_cbs { call_closure0(cb); } + let (finish_listeners, close_listeners) = listeners.unwrap_or_default(); + emit_no_arg_to_listeners(&finish_listeners); if callback != 0 { call_closure0(callback); } + emit_no_arg_to_listeners(&close_listeners); } /// Flush a standalone response: serialize the head + buffered body and diff --git a/crates/perry-ext-http/src/client_events.rs b/crates/perry-ext-http/src/client_events.rs index 4b8b6adca9..d18e03c3c2 100644 --- a/crates/perry-ext-http/src/client_events.rs +++ b/crates/perry-ext-http/src/client_events.rs @@ -1,6 +1,6 @@ -//! #4905 — client-request event helpers for the pending-event drain -//! loop: no-arg/error listener firing, transport-error → Node-coded -//! Error mapping, and the `'timeout'` event flow. +//! #4905 / #4909 — client-request event helpers for the pending-event +//! drain loop: response/error/timeout/flush handling, no-arg/error +//! listener firing, and transport-error → Node-coded Error mapping. use super::*; @@ -39,6 +39,26 @@ pub(crate) unsafe fn fire_request_error_listeners(request_handle: Handle, arg: f } } +/// Fire `'close'` exactly once per request (#4909 — the response, error, +/// timeout and destroy paths can each reach the close edge; Node emits it +/// a single time). +pub(crate) fn fire_request_close_once(request_handle: Handle) { + let fire = with_handle_mut::(request_handle, |req| { + if req.close_emitted { + false + } else { + req.close_emitted = true; + true + } + }) + .unwrap_or(false); + if fire { + unsafe { + fire_request_event_listeners(request_handle, "close"); + } + } +} + /// #4905 — map a transport error message to the value handed to /// `'error'` listeners. Recognized shapes become real Error objects /// carrying the Node `.code` (corpus tests assert @@ -67,46 +87,219 @@ pub(crate) fn error_event_arg(error_message: &str) -> f64 { } } -/// #4905 — drain handler for `PendingHttpEvent::Timeout`: fire -/// `'timeout'` listeners (falling back to the legacy error surface when -/// none are registered), honor an in-handler `req.destroy()` with the -/// coded ECONNRESET "socket hang up" error, then fire `'close'`. +/// Drain handler for `PendingHttpEvent::Response`: build the +/// IncomingMessage handle, call the factory callback and `'response'` +/// listeners, deliver `'data'`/`'end'`, then `'close'` on the request. +/// +/// # Safety +/// +/// Same listener-liveness contract as [`fire_request_event_listeners`]. +pub(crate) unsafe fn handle_response_event( + request_handle: Handle, + status: u16, + status_message: String, + headers: Vec<(String, String)>, + trailers: Vec<(String, String)>, + body: Vec, +) { + // #4909 — a destroyed request delivers nothing (Node tears the + // exchange down); `completed` also suppresses any late timeout timer. + let already_done = with_handle_mut::(request_handle, |req| { + let was = req.completed; + req.completed = true; + was + }) + .unwrap_or(false); + if already_done { + return; + } + + let response_callback = get_handle_mut::(request_handle) + .map(|r| r.response_callback) + .unwrap_or(0); + + let mut headers_map = HashMap::new(); + for (k, v) in headers { + headers_map.insert(k, v); + } + let mut trailers_map = HashMap::new(); + for (k, v) in trailers { + trailers_map.insert(k, v); + } + + let body_clone = body.clone(); + let incoming = register_handle(IncomingMessageHandle { + status_code: status, + status_message, + headers: headers_map, + trailers: trailers_map, + body, + listeners: HashMap::new(), + encoding: None, + }); + + // Hand the IncomingMessage handle to the user's `(res) => { ... }` + // callback. POINTER_TAG so the closure-arg unboxer extracts the i64. + let arg = f64::from_bits(POINTER_TAG | (incoming as u64 & PTR_MASK)); + if response_callback != 0 { + let closure = JsClosure::from_raw(response_callback as *const RawClosureHeader); + let _ = closure.call1(arg); + } + // #4909 — `.on('response', cb)` listeners fire too (the factory + // callback is just Node's pre-registered once-listener). + let response_listeners = get_handle_mut::(request_handle) + .and_then(|r| r.listeners.get("response").cloned()) + .unwrap_or_default(); + for cb in response_listeners { + if cb != 0 { + let closure = JsClosure::from_raw(cb as *const RawClosureHeader); + let _ = closure.call1(arg); + } + } + + // `'data'` listeners — body is delivered as a single chunk. True + // streaming requires a cooperative spawn_async perry-ffi surface + // (v0.6.0 followup). + // + // Issue #1124: bytes cross the FFI boundary as a JS Buffer + // (`alloc_buffer`), not a lossily-decoded string — unless + // `res.setEncoding(enc)` asked for Readable's string-chunk behavior. + let (data_listeners, encoding) = get_handle_mut::(incoming) + .map(|r| { + ( + r.listeners.get("data").cloned().unwrap_or_default(), + r.encoding.clone(), + ) + }) + .unwrap_or_default(); + if !data_listeners.is_empty() && !body_clone.is_empty() { + let arg = body_chunk_value(&body_clone, encoding.as_deref()); + if arg.to_bits() != TAG_UNDEFINED { + for cb in data_listeners { + if cb != 0 { + let closure = JsClosure::from_raw(cb as *const RawClosureHeader); + let _ = closure.call1(arg); + } + } + } + } + + // `'end'` listeners — fire after data. + let end_listeners = get_handle_mut::(incoming) + .and_then(|r| r.listeners.get("end").cloned()) + .unwrap_or_default(); + for cb in end_listeners { + if cb != 0 { + let closure = JsClosure::from_raw(cb as *const RawClosureHeader); + let _ = closure.call0(); + } + } + + // Node emits `'close'` on the request once the response has fully + // ended (#4905). + fire_request_close_once(request_handle); +} + +/// Drain handler for `PendingHttpEvent::Error`: `'error'` listeners then +/// `'close'`, suppressed entirely once the request already completed +/// (e.g. a `req.destroy()` raced the transport failure). +/// +/// # Safety +/// +/// Same listener-liveness contract as [`fire_request_event_listeners`]. +pub(crate) unsafe fn handle_error_event(request_handle: Handle, error_message: &str) { + let already_done = with_handle_mut::(request_handle, |req| { + let was = req.completed; + req.completed = true; + was + }) + .unwrap_or(false); + if already_done { + return; + } + fire_request_error_listeners(request_handle, error_event_arg(error_message)); + // Node emits `'close'` on the request after `'error'` (#4905). + fire_request_close_once(request_handle); +} + +/// #4905 / #4909 — drain handler for `PendingHttpEvent::Timeout`. +/// +/// `'timeout'` fires at most once per request and never after the +/// response/error completed it. For an in-flight exchange our transport +/// deadline has already aborted the request, so when nobody listens the +/// legacy error surface (+ `'close'`) keeps existing waiters finishing; +/// a request that was never dispatched just gets the event (Node doesn't +/// tear anything down on `'timeout'` — the canonical handler calls +/// `req.destroy()`, which emits its own coded ECONNRESET + `'close'`). /// /// # Safety /// /// Same listener-liveness contract as [`fire_request_event_listeners`]. pub(crate) unsafe fn handle_timeout_event(request_handle: Handle) { + let (fire, ended) = with_handle_mut::(request_handle, |req| { + if req.completed || req.timeout_fired { + (false, req.ended) + } else { + req.timeout_fired = true; + (true, req.ended) + } + }) + .unwrap_or((false, false)); + if !fire { + return; + } + let timeout_listeners = get_handle_mut::(request_handle) .and_then(|r| r.listeners.get("timeout").cloned()) .unwrap_or_default(); if timeout_listeners.is_empty() { - // No `'timeout'` listener — keep the legacy error surface. - fire_request_error_listeners(request_handle, error_event_arg("request timed out")); - } else { - for cb in timeout_listeners { - if cb != 0 { - let closure = JsClosure::from_raw(cb as *const RawClosureHeader); - let _ = closure.call0(); - } + if ended { + // In-flight exchange aborted by the transport deadline with no + // `'timeout'` listener — keep the legacy error surface. + fire_request_error_listeners(request_handle, error_event_arg("request timed out")); + fire_request_close_once(request_handle); } - // Node's `'timeout'` doesn't tear the request down by itself, but - // our transport deadline already aborted the exchange. The - // canonical pattern destroys the request in the handler and - // expects ECONNRESET "socket hang up"; honor that destroy with - // the coded error. - if client_request_surface::request_destroyed(request_handle) { - fire_request_error_listeners( - request_handle, - f64::from_bits( - perry_ffi::error_value_with_code( - "socket hang up", - "ECONNRESET", - perry_ffi::ErrorKind::Error, - ) - .bits(), - ), - ); + return; + } + for cb in timeout_listeners { + if cb != 0 { + let closure = JsClosure::from_raw(cb as *const RawClosureHeader); + let _ = closure.call0(); } } - fire_request_event_listeners(request_handle, "close"); + // The transport deadline killed an in-flight exchange; if the handler + // didn't destroy the request (destroy emits its own error + close), + // fire `'close'` so waiters still finish — nothing else will arrive. + if ended && !client_request_surface::request_destroyed(request_handle) { + fire_request_close_once(request_handle); + } +} + +/// #4909 — drain handler for `PendingHttpEvent::Flushed`: the body was +/// handed to the transport at `end()`. Node's flush ordering: queued +/// `write(chunk, cb)` callbacks (in order) → `'finish'` listeners → the +/// `end(..., cb)` callback. +/// +/// # Safety +/// +/// Same listener-liveness contract as [`fire_request_event_listeners`]. +pub(crate) unsafe fn handle_flushed_event(request_handle: Handle) { + let (write_cbs, end_cb) = with_handle_mut::(request_handle, |req| { + ( + std::mem::take(&mut req.pending_write_callbacks), + std::mem::replace(&mut req.end_callback, 0), + ) + }) + .unwrap_or_default(); + for cb in write_cbs { + if cb != 0 { + let closure = JsClosure::from_raw(cb as *const RawClosureHeader); + let _ = closure.call0(); + } + } + fire_request_event_listeners(request_handle, "finish"); + if end_cb != 0 { + let closure = JsClosure::from_raw(end_cb as *const RawClosureHeader); + let _ = closure.call0(); + } } diff --git a/crates/perry-ext-http/src/client_outgoing.rs b/crates/perry-ext-http/src/client_outgoing.rs new file mode 100644 index 0000000000..5b68fb936a --- /dev/null +++ b/crates/perry-ext-http/src/client_outgoing.rs @@ -0,0 +1,195 @@ +//! #4909 — client `OutgoingMessage` write/end callback + backpressure + +//! `setTimeout` surface. Mirrors the #4954 server-side fix +//! (`js_node_http_res_write_full` / `js_node_http_res_end_full`) on the +//! `http.ClientRequest` path: the static native-dispatch table routed +//! `req.write` / `req.end` to single-arg entry points that dropped the +//! `(encoding?, callback?)` tail, returned the handle from `write()` +//! (truthy forever, so `while (req.write(buf))` producer loops never +//! terminated), and silently discarded Buffer chunks. + +use super::*; + +/// Node's default `highWaterMark` for an HTTP `OutgoingMessage` (16 KiB). +/// `req.write()` returns `false` once the buffered body grows past this, +/// signalling backpressure so producer loops terminate. +const DEFAULT_HIGH_WATER_MARK: usize = 16 * 1024; + +extern "C" { + fn js_value_is_closure(value_bits: i64) -> i32; + fn js_buffer_is_buffer(ptr: i64) -> i32; +} + +/// Return the closure pointer carried by `value_bits` if it is a real +/// callable (POINTER_TAG + closure magic), else 0. `js_value_is_closure` +/// keeps a Buffer/object chunk — also POINTER_TAG — from being mistaken +/// for a callback. +pub(crate) fn callback_from_bits(value_bits: i64) -> i64 { + if unsafe { js_value_is_closure(value_bits) } != 0 { + (value_bits as u64 & PTR_MASK) as i64 + } else { + 0 + } +} + +/// Pick the callback from a `(encoding?, callback?)` trailing arg pair, +/// the later slot first — mirroring Node's `(chunk, encoding, callback)` +/// rule. A string encoding is not callable, so it is skipped. +pub(crate) fn pick_trailing_callback(arg2: i64, arg3: i64) -> i64 { + let c3 = callback_from_bits(arg3); + if c3 != 0 { + c3 + } else { + callback_from_bits(arg2) + } +} + +/// String / Buffer chunk → body bytes. Buffers must be probed first: a +/// Buffer is POINTER_TAG just like some heap strings, and +/// `extract_string_value` would misread its `BufferHeader` as a +/// `StringHeader` (the same layout confusion as #1124). +pub(crate) unsafe fn chunk_to_bytes(value: f64) -> Option> { + let bits = value.to_bits(); + if bits == TAG_UNDEFINED || bits == TAG_NULL { + return None; + } + if bits >> 48 == 0x7FFD { + let raw = (bits & PTR_MASK) as i64; + if js_buffer_is_buffer(raw) != 0 { + return perry_ffi::read_buffer_bytes(raw as *const perry_ffi::BufferHeader) + .map(|b| b.to_vec()); + } + } + extract_string_value(value).map(String::into_bytes) +} + +/// `req.write(chunk[, encoding][, callback])` — the full Node surface +/// routed from the static native dispatch table. The callback is queued +/// (it fires in order when the body flushes at `end()`); returns a +/// NaN-boxed boolean: `false` once the buffered body passes the 16 KiB +/// high-water mark (Node's backpressure signal), else `true`. +/// +/// # Safety +/// +/// FFI entry; `handle` must be a live `ClientRequestHandle` (or absent). +#[no_mangle] +pub unsafe extern "C" fn js_http_client_request_write_full( + handle: Handle, + chunk: f64, + arg2: i64, + arg3: i64, +) -> f64 { + let callback = pick_trailing_callback(arg2, arg3); + let bytes = chunk_to_bytes(chunk); + let mut below_hwm = true; + with_handle_mut::(handle, |req| { + if !req.ended { + if let Some(b) = &bytes { + req.body.extend_from_slice(b); + } + if callback != 0 { + req.pending_write_callbacks.push(callback); + } + below_hwm = req.body.len() <= DEFAULT_HIGH_WATER_MARK; + } + }); + f64::from_bits(if below_hwm { TAG_TRUE } else { TAG_FALSE }) +} + +/// `req.end([chunk][, encoding][, callback])` — the full Node surface +/// routed from the static native dispatch table. Handles the `end(cb)` +/// form (callback in the first slot) as well as +/// `end(chunk[, encoding][, callback])`. The queued write callbacks, the +/// `'finish'` listeners and the end callback fire on the next drain via +/// `PendingHttpEvent::Flushed` (queued by `client_request_end_impl`). +/// +/// # Safety +/// +/// FFI entry; `handle` must be a live `ClientRequestHandle` (or absent). +#[no_mangle] +pub unsafe extern "C" fn js_http_client_request_end_full( + handle: Handle, + chunk: f64, + arg2: i64, + arg3: i64, +) -> Handle { + let first_cb = callback_from_bits(chunk.to_bits() as i64); + let (real_chunk, callback) = if first_cb != 0 { + (f64::from_bits(TAG_UNDEFINED), first_cb) + } else { + (chunk, pick_trailing_callback(arg2, arg3)) + }; + if let Some(b) = chunk_to_bytes(real_chunk) { + with_handle_mut::(handle, |req| { + if !req.ended { + req.body.extend_from_slice(&b); + } + }); + } + if callback != 0 { + with_handle_mut::(handle, |req| { + if !req.ended { + req.end_callback = callback; + } + }); + } + client_request_end_impl(handle, f64::from_bits(TAG_UNDEFINED)) +} + +/// `req.setTimeout(msecs[, callback])` — the full Node surface. The +/// callback registers as a `'timeout'` listener, and a real timer is +/// armed immediately: Node's inactivity timer runs on the socket from +/// assignment, so `'timeout'` must fire even when the request was never +/// `end()`ed (the transport deadline alone only covered dispatched +/// requests — the canonical `req.setTimeout(n); req.on('timeout', …)` +/// pattern on a never-responding server hung forever, #4909). +/// +/// # Safety +/// +/// FFI entry; `handle` must be a live `ClientRequestHandle` (or absent). +#[no_mangle] +pub unsafe extern "C" fn js_http_set_timeout_full( + handle: Handle, + ms: f64, + callback_bits: i64, +) -> Handle { + let cb = callback_from_bits(callback_bits); + if cb != 0 { + with_handle_mut::(handle, |req| { + req.listeners + .entry("timeout".to_string()) + .or_default() + .push(cb); + }); + } + client_request_set_timeout_impl(handle, ms); + let effective = + with_handle_mut::(handle, |req| req.timeout_ms).flatten(); + if let Some(ms) = effective { + arm_client_timeout(handle, ms); + } + handle +} + +/// Arm a one-shot timer that pushes `PendingHttpEvent::Timeout` after +/// `ms` milliseconds. The drain dedupes (`timeout_fired`) and suppresses +/// stale timers (`completed`), so over-arming is harmless — rescheduled +/// `setTimeout` calls and the per-dispatch transport deadline can all +/// race the same request safely. +pub(crate) fn arm_client_timeout(request_handle: Handle, ms: u64) { + spawn_blocking(move || { + // Defeat LTO dead-stripping of tokio's CONTEXT statics — same + // workaround dispatch_request needs (see spawn_socket_runner). + let try_h = tokio::runtime::Handle::try_current(); + std::hint::black_box(&try_h); + if try_h.is_err() { + return; + } + let handle = tokio::runtime::Handle::current(); + let jh = handle.spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(ms)).await; + push_event(PendingHttpEvent::Timeout { request_handle }); + }); + std::hint::black_box(&jh); + std::mem::forget(jh); + }); +} diff --git a/crates/perry-ext-http/src/client_request_surface.rs b/crates/perry-ext-http/src/client_request_surface.rs index 86d1e63edd..7dcdce3622 100644 --- a/crates/perry-ext-http/src/client_request_surface.rs +++ b/crates/perry-ext-http/src/client_request_surface.rs @@ -153,6 +153,20 @@ fn headers_object(handle: Handle) -> f64 { f64::from_bits(perry_runtime::JSValue::object_ptr(obj as *mut u8).bits()) } +/// `{ name: }` — stands in for `.constructor` so +/// `out.constructor.name` discriminates ClientRequest/ServerResponse the +/// way the corpus outgoing-message tests expect (#4909). +pub(crate) fn constructor_object(name: &str) -> f64 { + let obj = perry_runtime::js_object_alloc_null_proto(0, 1); + let key_ptr = perry_runtime::js_string_from_bytes("name".as_ptr(), 4); + let value_ptr = perry_runtime::js_string_from_bytes(name.as_ptr(), name.len() as u32); + perry_runtime::js_object_set_field(obj, 0, perry_runtime::JSValue::string_ptr(value_ptr)); + let mut keys = perry_runtime::js_array_alloc(1); + keys = perry_runtime::js_array_push(keys, perry_runtime::JSValue::string_ptr(key_ptr)); + perry_runtime::js_object_set_keys(obj, keys); + f64::from_bits(perry_runtime::JSValue::object_ptr(obj as *mut u8).bits()) +} + fn socket_value(handle: Handle) -> f64 { if !is_client_request_handle(handle) { return undefined_value(); @@ -258,9 +272,41 @@ pub extern "C" fn js_http_client_request_abort(handle: Handle) -> f64 { #[no_mangle] pub extern "C" fn js_http_client_request_destroy(handle: Handle, _error: f64) -> Handle { - if is_client_request_handle(handle) { - with_state_mut(handle, |state| state.destroyed = true); + if !is_client_request_handle(handle) { + return handle; + } + let already = with_state_mut(handle, |state| { + std::mem::replace(&mut state.destroyed, true) + }); + if already { + return handle; + } + // #4909 — Node teardown: destroying an in-flight request (sent, no + // response yet) emits the coded ECONNRESET "socket hang up" on + // `'error'`, then `'close'`. A request that never went out (or whose + // response already completed) just gets the (once-only) `'close'`. + let in_flight = with_handle_mut::(handle, |req| { + let was_completed = req.completed; + req.completed = true; + req.ended && !was_completed + }) + .unwrap_or(false); + if in_flight { + unsafe { + client_events::fire_request_error_listeners( + handle, + f64::from_bits( + perry_ffi::error_value_with_code( + "socket hang up", + "ECONNRESET", + perry_ffi::ErrorKind::Error, + ) + .bits(), + ), + ); + } } + client_events::fire_request_close_once(handle); handle } @@ -321,6 +367,12 @@ fn dispatch_property(handle: Handle, property: &str) -> Option { } let method: Option<&'static [u8]> = match property { "on" => Some(b"on"), + "once" => Some(b"once"), + "addListener" => Some(b"addListener"), + "prependListener" => Some(b"prependListener"), + "removeListener" => Some(b"removeListener"), + "off" => Some(b"off"), + "removeAllListeners" => Some(b"removeAllListeners"), "end" => Some(b"end"), "write" => Some(b"write"), "setHeader" => Some(b"setHeader"), @@ -382,6 +434,10 @@ fn dispatch_property(handle: Handle, property: &str) -> Option { string_value(&path) }) .unwrap_or_else(undefined_value), + // #4909 — `out.constructor.name` discrimination (the corpus + // outgoing-message tests branch on it). A plain `{ name }` object: + // the real class object isn't reachable from a raw handle. + "constructor" => constructor_object("ClientRequest"), "aborted" => js_http_client_request_aborted(handle), "destroyed" => js_http_client_request_destroyed(handle), "finished" => js_http_client_request_finished(handle), @@ -398,25 +454,33 @@ fn dispatch_method(handle: Handle, method: &str, args: &[f64]) -> Option { if !is_client_request_handle(handle) { return None; } + // #4909 — the `(encoding?, callback?)` tail rides in args[1]/args[2] + // as raw NaN-boxed bits for the write/end/setTimeout surfaces. + let arg_bits = |index: usize| -> i64 { + args.get(index) + .map(|v| v.to_bits() as i64) + .unwrap_or(TAG_UNDEFINED as i64) + }; Some(match method { "end" => { unsafe { - client_request_end_impl( - handle, - args.first().copied().unwrap_or_else(undefined_value), - ); - } - handle_value(handle) - } - "write" => { - unsafe { - client_request_write_impl( + client_outgoing::js_http_client_request_end_full( handle, args.first().copied().unwrap_or_else(undefined_value), + arg_bits(1), + arg_bits(2), ); } handle_value(handle) } + "write" => unsafe { + client_outgoing::js_http_client_request_write_full( + handle, + args.first().copied().unwrap_or_else(undefined_value), + arg_bits(1), + arg_bits(2), + ) + }, "setHeader" => { let name = string_arg(args, 0).unwrap_or_default(); let value = string_arg(args, 1).unwrap_or_default(); @@ -443,7 +507,11 @@ fn dispatch_method(handle: Handle, method: &str, args: &[f64]) -> Option { "getRawHeaderNames" => headers_array(handle, true), "setTimeout" => { unsafe { - client_request_set_timeout_impl(handle, args.first().copied().unwrap_or(0.0)); + client_outgoing::js_http_set_timeout_full( + handle, + args.first().copied().unwrap_or(0.0), + arg_bits(1), + ); } handle_value(handle) } @@ -463,6 +531,44 @@ fn dispatch_method(handle: Handle, method: &str, args: &[f64]) -> Option { } "abort" => js_http_client_request_abort(handle), "destroy" => handle_value(js_http_client_request_destroy(handle, undefined_value())), + // #4909 — the dynamic path (an untyped `out` parameter) used to fall + // through to the unknown-method arm here, so `.on('finish', cb)` + // silently dropped the listener even though the statically-typed + // route registered it fine via `js_http_on`. + "on" | "once" | "addListener" | "prependListener" => { + let event = string_arg(args, 0).unwrap_or_default(); + let cb = client_outgoing::callback_from_bits(arg_bits(1)); + if !event.is_empty() && cb != 0 { + with_handle_mut::(handle, |req| { + req.listeners.entry(event.clone()).or_default().push(cb); + }); + } + handle_value(handle) + } + "removeListener" | "off" => { + let event = string_arg(args, 0).unwrap_or_default(); + let cb = client_outgoing::callback_from_bits(arg_bits(1)); + if !event.is_empty() && cb != 0 { + with_handle_mut::(handle, |req| { + if let Some(cbs) = req.listeners.get_mut(&event) { + if let Some(pos) = cbs.iter().position(|&c| c == cb) { + cbs.remove(pos); + } + } + }); + } + handle_value(handle) + } + "removeAllListeners" => { + let event = string_arg(args, 0); + with_handle_mut::(handle, |req| match &event { + Some(e) => { + req.listeners.remove(e); + } + None => req.listeners.clear(), + }); + handle_value(handle) + } "flushHeaders" | "cork" | "uncork" | "setNoDelay" | "setSocketKeepAlive" => { undefined_value() } diff --git a/crates/perry-ext-http/src/lib.rs b/crates/perry-ext-http/src/lib.rs index bc3a4d52ab..f310f7b045 100644 --- a/crates/perry-ext-http/src/lib.rs +++ b/crates/perry-ext-http/src/lib.rs @@ -68,7 +68,10 @@ use client_dispatch::dispatch_request; // Client-request event drain helpers (#4905) — extracted from this file // to stay under the 2000-line lint cap. mod client_events; -use client_events::{error_event_arg, fire_request_error_listeners, fire_request_event_listeners}; + +// Client OutgoingMessage write/end callback + backpressure + setTimeout +// surface (#4909) — extracted to stay under the 2000-line lint cap. +mod client_outgoing; // Node-compatible argument/header/URL validation for the client factories // (#4907) — throws `ERR_*`-coded errors on bad input. @@ -91,6 +94,8 @@ const POINTER_TAG: u64 = 0x7FFD_0000_0000_0000; const PTR_MASK: u64 = 0x0000_FFFF_FFFF_FFFF; const TAG_UNDEFINED: u64 = 0x7FFC_0000_0000_0001; const TAG_NULL: u64 = 0x7FFC_0000_0000_0002; +const TAG_FALSE: u64 = 0x7FFC_0000_0000_0003; +const TAG_TRUE: u64 = 0x7FFC_0000_0000_0004; // ------------------------------------------------------------------ // Pending event queue + GC scanner @@ -115,6 +120,10 @@ pub(crate) enum PendingHttpEvent { /// listeners when any exist; falls back to the Error surface /// otherwise. Timeout { request_handle: Handle }, + /// #4909 — the request body was handed to the transport at `end()`. + /// Drains the queued `write(chunk, cb)` callbacks, then `'finish'`, + /// then the `end(..., cb)` callback — Node's flush ordering. + Flushed { request_handle: Handle }, } lazy_static! { @@ -165,6 +174,10 @@ pub(crate) fn ensure_gc_scanner_registered() { fn scan_http_roots(visitor: &mut GcRootVisitor<'_>) { iter_handles_of_mut::(|req| { visitor.visit_i64_slot(&mut req.response_callback); + visitor.visit_i64_slot(&mut req.end_callback); + for cb in &mut req.pending_write_callbacks { + visitor.visit_i64_slot(cb); + } for cbs in req.listeners.values_mut() { for cb in cbs { visitor.visit_i64_slot(cb); @@ -433,10 +446,26 @@ pub struct ClientRequestHandle { headers: HashMap, body: Vec, response_callback: i64, - /// `'error'` is the only event ClientRequest emits today. + /// `.on(event, cb)` listeners (`'response'` / `'error'` / `'timeout'` + /// / `'finish'` / `'close'`). listeners: HashMap>, timeout_ms: Option, ended: bool, + /// #4909 — `write(chunk, cb)` callbacks queued until the body is + /// flushed at `end()` (Node fires them once the chunk hits the + /// transport; our buffered MVP flushes everything at `end()`). + pending_write_callbacks: Vec, + /// #4909 — the `end(..., cb)` callback; fires after the queued write + /// callbacks and the `'finish'` listeners. + end_callback: i64, + /// #4909 — set once the response/error was delivered (or the request + /// destroyed); suppresses late `'timeout'` timers and stale events. + completed: bool, + /// #4909 — `'timeout'` fires at most once per request, no matter how + /// many timers (`options.timeout` + `setTimeout()` reschedules) land. + timeout_fired: bool, + /// #4909 — `'close'` fires at most once per request. + close_emitted: bool, /// `options.agent` handle id when the caller supplied an Agent /// (#2154). `0` = use the global `HTTP_CLIENT` (no pooling /// distinction). When set, `dispatch_request` calls @@ -594,7 +623,7 @@ fn make_request_handle( callback: i64, agent_handle: Handle, ) -> Handle { - register_handle(ClientRequestHandle { + let handle = register_handle(ClientRequestHandle { method, url, headers, @@ -603,9 +632,23 @@ fn make_request_handle( listeners: HashMap::new(), timeout_ms, ended: false, + pending_write_callbacks: Vec::new(), + end_callback: 0, + completed: false, + timeout_fired: false, + close_emitted: false, agent_handle, tls: tls_client::TlsOptions::default(), - }) + }); + // #4909 — `options.timeout` arms the inactivity timer as soon as the + // socket exists in Node, not at `end()`; a request that is never + // dispatched (or whose server never answers) still gets `'timeout'`. + if let Some(ms) = timeout_ms { + if ms > 0 { + client_outgoing::arm_client_timeout(handle, ms); + } + } + handle } /// Parse the client-side TLS options (#4906) off a request options value @@ -1090,9 +1133,11 @@ pub unsafe extern "C" fn js_http_client_request_write(handle: Handle, body_f64: } unsafe fn client_request_write_impl(handle: Handle, body_f64: f64) -> Handle { - if let Some(body) = extract_string_value(body_f64) { + // #4909 — Buffer chunks used to be misread as StringHeaders (and + // dropped); route through the buffer-aware chunk reader. + if let Some(body) = client_outgoing::chunk_to_bytes(body_f64) { with_handle_mut::(handle, |req| { - req.body.extend_from_slice(body.as_bytes()); + req.body.extend_from_slice(&body); }); } handle @@ -1106,10 +1151,10 @@ pub unsafe extern "C" fn js_http_client_request_end(handle: Handle, body_f64: f6 client_request_end_impl(handle, body_f64) } -unsafe fn client_request_end_impl(handle: Handle, body_f64: f64) -> Handle { - if let Some(body) = extract_string_value(body_f64) { +pub(crate) unsafe fn client_request_end_impl(handle: Handle, body_f64: f64) -> Handle { + if let Some(body) = client_outgoing::chunk_to_bytes(body_f64) { with_handle_mut::(handle, |req| { - req.body.extend_from_slice(body.as_bytes()); + req.body.extend_from_slice(&body); }); } @@ -1136,6 +1181,12 @@ unsafe fn client_request_end_impl(handle: Handle, body_f64: f64) -> Handle { let (method, url, headers, body, timeout_ms, agent_handle, tls) = snapshot; + // #4909 — queue the flush notification before dispatching so the + // write/end callbacks and `'finish'` drain ahead of any `'response'`. + push_event(PendingHttpEvent::Flushed { + request_handle: handle, + }); + // #2154 — if the agent supplied a `createConnection` / `createSocket` // override, invoke it here on the main thread (JS closure calls must not // run on a tokio worker) and run the HTTP exchange over the socket it @@ -1264,7 +1315,7 @@ pub unsafe extern "C" fn js_http_set_timeout(handle: Handle, ms: f64) -> Handle client_request_set_timeout_impl(handle, ms) } -unsafe fn client_request_set_timeout_impl(handle: Handle, ms: f64) -> Handle { +pub(crate) unsafe fn client_request_set_timeout_impl(handle: Handle, ms: f64) -> Handle { // Node's `socket.setTimeout` (which backs `ClientRequest.setTimeout`) // routes the delay through validateTimerDuration → enroll: an out-of-range // (> 2**31-1) delay is clamped to TIMEOUT_MAX and a `TimeoutOverflowWarning` @@ -1278,7 +1329,12 @@ unsafe fn client_request_set_timeout_impl(handle: Handle, ms: f64) -> Handle { ms }; with_handle_mut::(handle, |req| { - req.timeout_ms = Some(effective.max(0.0) as u64); + // Node: `setTimeout(0)` clears the inactivity timer. + req.timeout_ms = if effective > 0.0 { + Some(effective as u64) + } else { + None + }; }); handle } @@ -1523,7 +1579,7 @@ fn server_incoming_property(handle: Handle, property_name: &str) -> Option } } -fn body_chunk_value(body: &[u8], encoding: Option<&str>) -> f64 { +pub(crate) fn body_chunk_value(body: &[u8], encoding: Option<&str>) -> f64 { match encoding { Some(_) => { let s = String::from_utf8_lossy(body).into_owned(); @@ -1584,107 +1640,27 @@ pub unsafe extern "C" fn js_http_process_pending() -> i32 { trailers, body, } => { - let response_callback = get_handle_mut::(request_handle) - .map(|r| r.response_callback) - .unwrap_or(0); - - let mut headers_map = HashMap::new(); - for (k, v) in headers { - headers_map.insert(k, v); - } - let mut trailers_map = HashMap::new(); - for (k, v) in trailers { - trailers_map.insert(k, v); - } - - let body_clone = body.clone(); - let incoming = register_handle(IncomingMessageHandle { - status_code: status, + client_events::handle_response_event( + request_handle, + status, status_message, - headers: headers_map, - trailers: trailers_map, + headers, + trailers, body, - listeners: HashMap::new(), - encoding: None, - }); - - if response_callback != 0 { - // Hand the IncomingMessage handle to the user's - // `(res) => { ... }` callback. POINTER_TAG so the - // closure-arg unboxer extracts the i64. - let arg = f64::from_bits(POINTER_TAG | (incoming as u64 & PTR_MASK)); - let closure = JsClosure::from_raw(response_callback as *const RawClosureHeader); - let _ = closure.call1(arg); - } - - // `'data'` listeners — body is delivered as a single chunk. - // True streaming requires a cooperative spawn_async - // perry-ffi surface (v0.6.0 followup). - // - // Issue #1124 followup: pre-fix this allocated a JS - // string via `alloc_string(str::from_utf8(&body).unwrap_or(""))`, - // which silently collapsed any non-UTF-8 byte sequence - // (PNG file-magic, gzip frames, binary protocols, …) to - // the empty string before user code ever saw a byte. - // The mirror of the #1124 server-side fix (where the - // request body went the OTHER direction through a - // wrongly-shaped StringHeader): allocate a JS Buffer - // via `alloc_buffer(&bytes)` so the bytes survive the - // FFI boundary intact. The Buffer registers itself - // through perry-runtime's `is_registered_buffer` path - // so the `chunk.toString(enc)` / `chunk.length` / - // `Buffer.concat(...)` surface lights up on the - // returned value. - // - // When `res.setEncoding(enc)` was called in the response - // callback, mirror Readable's string-chunk behavior. Without - // an encoding, preserve Node's default Buffer chunks. - let (data_listeners, encoding) = get_handle_mut::(incoming) - .map(|r| { - ( - r.listeners.get("data").cloned().unwrap_or_default(), - r.encoding.clone(), - ) - }) - .unwrap_or_default(); - if !data_listeners.is_empty() && !body_clone.is_empty() { - let arg = body_chunk_value(&body_clone, encoding.as_deref()); - if arg.to_bits() != TAG_UNDEFINED { - for cb in data_listeners { - if cb != 0 { - let closure = JsClosure::from_raw(cb as *const RawClosureHeader); - let _ = closure.call1(arg); - } - } - } - } - - // `'end'` listeners — fire after data. - let end_listeners = get_handle_mut::(incoming) - .and_then(|r| r.listeners.get("end").cloned()) - .unwrap_or_default(); - for cb in end_listeners { - if cb != 0 { - let closure = JsClosure::from_raw(cb as *const RawClosureHeader); - let _ = closure.call0(); - } - } - - // Node emits `'close'` on the request once the response has - // fully ended (#4905). - fire_request_event_listeners(request_handle, "close"); + ); } PendingHttpEvent::Error { request_handle, error_message, } => { - fire_request_error_listeners(request_handle, error_event_arg(&error_message)); - // Node emits `'close'` on the request after `'error'` (#4905). - fire_request_event_listeners(request_handle, "close"); + client_events::handle_error_event(request_handle, &error_message); } PendingHttpEvent::Timeout { request_handle } => { client_events::handle_timeout_event(request_handle); } + PendingHttpEvent::Flushed { request_handle } => { + client_events::handle_flushed_event(request_handle); + } } } diff --git a/crates/perry-ext-http/src/tests.rs b/crates/perry-ext-http/src/tests.rs index 0e5817ed3d..4f18d7f48f 100644 --- a/crates/perry-ext-http/src/tests.rs +++ b/crates/perry-ext-http/src/tests.rs @@ -65,6 +65,11 @@ fn gc_mutable_scanner_rewrites_request_response_listener_roots() { listeners: request_listeners, timeout_ms: None, ended: false, + pending_write_callbacks: Vec::new(), + end_callback: 0, + completed: false, + timeout_fired: false, + close_emitted: false, agent_handle: 0, tls: crate::tls_client::TlsOptions::default(), }); diff --git a/crates/perry-stdlib/src/common/dispatch.rs b/crates/perry-stdlib/src/common/dispatch.rs index cf512642ca..4df8736d97 100644 --- a/crates/perry-stdlib/src/common/dispatch.rs +++ b/crates/perry-stdlib/src/common/dispatch.rs @@ -1945,6 +1945,7 @@ pub unsafe extern "C" fn js_handle_property_dispatch( | "resume" | "destroy" | "read" + | "constructor" ) && unsafe { js_ext_http_incoming_message_is_handle(handle) } != 0 { return unsafe { @@ -1997,6 +1998,7 @@ pub unsafe extern "C" fn js_handle_property_dispatch( | "writeProcessing" | "on" | "addListener" + | "constructor" ) && unsafe { js_ext_http_server_response_is_handle(handle) } != 0 { return unsafe { diff --git a/crates/perry-stdlib/src/common/dispatch_http.rs b/crates/perry-stdlib/src/common/dispatch_http.rs index 4bc83f43e2..fd30a8a85d 100644 --- a/crates/perry-stdlib/src/common/dispatch_http.rs +++ b/crates/perry-stdlib/src/common/dispatch_http.rs @@ -27,6 +27,16 @@ pub(super) unsafe fn dispatch_client_request_method( | "uncork" | "setNoDelay" | "setSocketKeepAlive" + // #4909 — listener management: `.on('finish'|'timeout'|…, cb)` + // on an untyped receiver silently dropped the listener (the + // ext dispatcher handles these now; the gate hid them). + | "on" + | "once" + | "addListener" + | "prependListener" + | "removeListener" + | "off" + | "removeAllListeners" ) { return None; } @@ -95,6 +105,14 @@ pub(super) unsafe fn dispatch_client_request_property( | "writableFinished" | "socket" | "connection" + // #4909 — listener management binds + `constructor.name`. + | "once" + | "addListener" + | "prependListener" + | "removeListener" + | "off" + | "removeAllListeners" + | "constructor" ) { return None; } diff --git a/crates/perry/tests/issue_4909_client_write_end_timeouts.rs b/crates/perry/tests/issue_4909_client_write_end_timeouts.rs new file mode 100644 index 0000000000..05dceaf6a9 --- /dev/null +++ b/crates/perry/tests/issue_4909_client_write_end_timeouts.rs @@ -0,0 +1,155 @@ +//! Regression tests for the #4909 client-side OutgoingMessage + timeout +//! sub-tickets: `req.write(chunk, cb)` / `req.end(cb)` fire their callbacks +//! in Node's flush order (write callbacks → `'finish'` → end callback), +//! `req.write()` returns a real backpressure boolean, and +//! `req.setTimeout(ms, cb)` emits `'timeout'` even when the server never +//! responds — followed by the coded ECONNRESET + `'close'` teardown when the +//! handler destroys the request. +//! +//! Pre-fix, the static dispatch table routed `req.write`/`req.end` to +//! single-arg entry points that dropped the callbacks and returned the +//! (always-truthy) handle from `write()`, so `while (req.write(buf, cb))` +//! producer loops spun forever; `req.setTimeout()` only stored the delay, +//! so a never-responding server hung the process — the dominant silent-hang +//! mode in the #4909 corpus bucket. + +use std::path::PathBuf; +use std::process::Command; + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +fn compile_and_run(dir: &std::path::Path, source: &str) -> String { + let entry = dir.join("main.ts"); + let output = dir.join("main_bin"); + std::fs::write(&entry, source).expect("write entry"); + + let compile = Command::new(perry_bin()) + .current_dir(dir) + .arg("compile") + .arg(&entry) + .arg("-o") + .arg(&output) + .output() + .expect("run perry compile"); + assert!( + compile.status.success(), + "perry compile failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&compile.stdout), + String::from_utf8_lossy(&compile.stderr) + ); + + let run = Command::new(&output) + .current_dir(dir) + .output() + .expect("run compiled binary"); + assert!( + run.status.success(), + "compiled binary failed\nstatus: {:?}\nstdout:\n{}\nstderr:\n{}", + run.status, + String::from_utf8_lossy(&run.stdout), + String::from_utf8_lossy(&run.stderr) + ); + String::from_utf8_lossy(&run.stdout).into_owned() +} + +/// Asserts `needles` appear in `haystack` in the given order. +fn assert_ordered(haystack: &str, needles: &[&str]) { + let mut from = 0; + for needle in needles { + match haystack[from..].find(needle) { + Some(pos) => from += pos + needle.len(), + None => panic!("expected \"{needle}\" (in order) in output:\n{haystack}"), + } + } +} + +/// Client + server write/end callbacks, backpressure boolean, and Node's +/// flush ordering (write cbs → 'finish' → end cb) on both OutgoingMessage +/// directions. +#[test] +fn write_end_callbacks_fire_in_flush_order_with_backpressure() { + let dir = tempfile::tempdir().expect("tempdir"); + let stdout = compile_and_run( + dir.path(), + r#" +const http = require('http'); +const server = http.createServer((req: any, res: any) => { + req.resume(); + req.on('end', () => { + res.write('x', () => console.log('res-wcb')); + res.on('finish', () => console.log('res-finish')); + res.end('y', () => console.log('res-endcb')); + }); + server.close(); +}).listen(0, function () { + const req = http.request({ port: server.address().port, method: 'PUT' }); + const big = Buffer.alloc(16 * 1024, 'x'); + // First 16 KiB write sits exactly at the high-water mark (true), the + // second one passes it (false) — the Node backpressure contract that + // terminates `while (req.write(buf))` producer loops. + console.log('w1', req.write(big, () => console.log('req-wcb1'))); + console.log('w2', req.write(big, () => console.log('req-wcb2'))); + req.on('finish', () => console.log('req-finish')); + req.end(() => console.log('req-endcb')); + req.on('response', (res: any) => { console.log('status', res.statusCode); }); +}); +"#, + ); + + assert_ordered( + &stdout, + &[ + "w1 true", + "w2 false", + "req-wcb1", + "req-wcb2", + "req-finish", + "req-endcb", + "res-wcb", + "res-finish", + "res-endcb", + "status 200", + ], + ); +} + +/// `req.setTimeout(ms, cb)` against a server that never responds: the +/// `'timeout'` event must fire (pre-fix the delay was stored but no timer +/// existed, hanging forever), and the canonical destroy-in-handler pattern +/// gets the coded ECONNRESET "socket hang up" then `'close'` with +/// `req.destroyed === true`. +#[test] +fn set_timeout_fires_timeout_event_and_destroy_tears_down() { + let dir = tempfile::tempdir().expect("tempdir"); + let stdout = compile_and_run( + dir.path(), + r#" +const http = require('http'); +const server = http.createServer(() => { /* never respond */ }); +server.listen(0, function () { + const req = http.request({ port: server.address().port }); + req.setTimeout(50, () => { + console.log('timeout-fired'); + req.destroy(); + }); + req.on('error', (e: any) => console.log('error-code', e.code)); + req.on('close', () => { + console.log('close-destroyed', req.destroyed); + server.close(); + }); + req.end(); +}); +"#, + ); + + assert_ordered( + &stdout, + &[ + "timeout-fired", + "error-code ECONNRESET", + "close-destroyed true", + ], + ); +}