fix(http): client write/end callbacks + backpressure, real client timeouts, dynamic listener registration (#4909)

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")),

crates/perry-codegen/src/lower_call/native_table/http_client.rs

@@ -141,18 +141,28 @@ 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 {
         module: "http",
         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 {

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]);

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: <class name> }` — stands in for `<handle>.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<f64>) -> 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
 }
 

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::<IncomingMessage>(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;
         }

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::<ServerResponse>(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::<ServerResponse>(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