diff --git a/crates/perry-codegen/src/lower_call/native_table/net_events.rs b/crates/perry-codegen/src/lower_call/native_table/net_events.rs index c784f97157..0918b10364 100644 --- a/crates/perry-codegen/src/lower_call/native_table/net_events.rs +++ b/crates/perry-codegen/src/lower_call/native_table/net_events.rs @@ -299,7 +299,9 @@ pub(super) const NET_EVENTS_ROWS: &[NativeModSig] = &[ method: "connect", class_filter: Some("Socket"), runtime: "js_net_socket_method_connect", - args: &[NA_F64, NA_STR], + // Keep every slot raw so port/host, options, and path overloads reach + // the runtime without callback-to-string coercion. + args: &[NA_F64, NA_F64, NA_F64], ret: NR_VOID, }, NativeModSig { diff --git a/crates/perry-ext-net/src/dispatch.rs b/crates/perry-ext-net/src/dispatch.rs index 8909755f15..76afc8fbe8 100644 --- a/crates/perry-ext-net/src/dispatch.rs +++ b/crates/perry-ext-net/src/dispatch.rs @@ -232,9 +232,11 @@ unsafe fn socket_method(handle: i64, method: &str, args: &[f64]) -> Option crate::js_net_socket_on(handle, unbox_to_i64(args[0]), unbox_to_i64(args[1])); nanbox_handle(handle) } - "connect" if args.len() >= 2 => { - crate::js_net_socket_method_connect(handle, args[0], unbox_to_i64(args[1])); - undefined() + "connect" if !args.is_empty() => { + let arg2 = args.get(1).copied().unwrap_or_else(undefined); + let arg3 = args.get(2).copied().unwrap_or_else(undefined); + crate::js_ext_net_socket_method_connect(handle, args[0], arg2, arg3); + nanbox_handle(handle) } "upgradeToTLS" if !args.is_empty() => { let verify = args.get(1).copied().unwrap_or(1.0); diff --git a/crates/perry-ext-net/src/ipc.rs b/crates/perry-ext-net/src/ipc.rs new file mode 100644 index 0000000000..a13eb49a53 --- /dev/null +++ b/crates/perry-ext-net/src/ipc.rs @@ -0,0 +1,407 @@ +//! Local IPC transport for `node:net` path overloads. +//! +//! Node maps `server.listen(path)` and `net.connect({ path })` to named pipes +//! on Windows and Unix-domain sockets on Unix. The streams join the same +//! SocketState command/event loop as TCP, so data, end, error, close, and +//! server connection events keep one implementation. + +use std::io; + +#[cfg(windows)] +use std::time::Duration; + +use tokio::sync::{mpsc, oneshot}; + +use crate::{ + dispatch, ensure_gc_scanner_registered, mark_closed, next_id, next_id_or_throw, push_event, + run_socket_task, server_state, statics, PendingNetEvent, SocketCommand, SocketState, + TlsSocketMetadata, Transport, +}; + +#[cfg(unix)] +use tokio::net::{UnixListener, UnixStream}; + +#[cfg(windows)] +use tokio::net::windows::named_pipe::{ClientOptions, NamedPipeServer, ServerOptions}; + +fn allocate_socket() -> (i64, mpsc::UnboundedReceiver) { + ensure_gc_scanner_registered(); + dispatch::ensure_runtime_dispatch_registered(); + let id = next_id_or_throw(); + let (tx, rx) = mpsc::unbounded_channel::(); + statics::sockets().lock().unwrap().insert( + id, + SocketState { + cmd_tx: tx, + pending_rx: None, + is_open: false, + refed: true, + local_addr: None, + raw: None, + destroyed: false, + bytes_read: 0, + bytes_written: 0, + timeout: None, + type_of_service: 0, + server_id: None, + server_connection_active: false, + tls: TlsSocketMetadata::default(), + }, + ); + statics::listeners() + .lock() + .unwrap() + .insert(id, Default::default()); + (id, rx) +} + +/// Read a JS string without coercing closures or option objects through a +/// StringHeader layout. +pub(crate) unsafe fn string_value(value: f64) -> Option { + let value = perry_ffi::JsValue::from_bits(value.to_bits()); + value + .is_string() + .then(|| crate::string_from_header_i64(value.as_string_ptr() as i64))? +} + +pub(crate) fn register_connect_cb(handle: i64, cb_f64: f64) { + if handle == 0 || !crate::is_nanboxed_pointer(cb_f64) { + return; + } + let cb_ptr = unsafe { crate::unbox_pointer(cb_f64) } as i64; + if cb_ptr == 0 { + return; + } + statics::listeners() + .lock() + .unwrap() + .entry(handle) + .or_default() + .entry("connect".to_string()) + .or_default() + .push(cb_ptr); +} + +/// Publish an accepted TCP or IPC stream as a normal net.Socket and start its +/// shared command/read loop. Admission accounting has already reserved one +/// pending connection before this helper is called. +pub(crate) fn register_accepted_transport( + server_id: i64, + transport: Transport, + local_addr: Option, +) { + let socket_id = next_id(); + if socket_id == perry_ffi::INVALID_HANDLE { + server_state::cancel_pending_connection(server_id); + return; + } + let (tx, rx) = mpsc::unbounded_channel::(); + statics::sockets().lock().unwrap().insert( + socket_id, + SocketState { + cmd_tx: tx, + pending_rx: None, + is_open: true, + refed: true, + local_addr, + raw: None, + destroyed: false, + bytes_read: 0, + bytes_written: 0, + timeout: None, + type_of_service: 0, + server_id: Some(server_id), + server_connection_active: false, + tls: TlsSocketMetadata::default(), + }, + ); + statics::listeners() + .lock() + .unwrap() + .insert(socket_id, Default::default()); + push_event(PendingNetEvent::ServerConnection( + server_id, socket_id, false, + )); + tokio::spawn(async move { + let mut rx = rx; + run_socket_task(socket_id, transport, &mut rx).await; + }); +} + +pub(crate) fn spawn_socket(path: String) -> i64 { + let (id, rx) = allocate_socket(); + spawn_connect(id, path, rx); + id +} + +pub(crate) fn connect_existing(handle: i64, path: String) { + let rx = { + let mut sockets = statics::sockets().lock().unwrap(); + match sockets + .get_mut(&handle) + .and_then(|socket| socket.pending_rx.take()) + { + Some(rx) => rx, + None => { + push_event(PendingNetEvent::Error( + handle, + "socket already connected (or unknown handle)".to_string(), + )); + return; + } + } + }; + spawn_connect(handle, path, rx); +} + +fn spawn_connect(id: i64, path: String, mut rx: mpsc::UnboundedReceiver) { + let local_server = server_state::begin_local_path_connect(&path); + crate::spawn_socket_runner(move || { + Box::pin(async move { + let stream = match connect_path(&path).await { + Ok(stream) => stream, + Err(error) => { + server_state::cancel_local_connect(local_server); + push_event(PendingNetEvent::Error( + id, + format!("connect {path}: {error}"), + )); + push_event(PendingNetEvent::Close(id)); + mark_closed(id); + return; + } + }; + + if let Some(socket) = statics::sockets().lock().unwrap().get_mut(&id) { + socket.is_open = true; + } + tokio::task::yield_now().await; + push_event(PendingNetEvent::Connect(id, local_server)); + run_socket_task(id, Transport::Ipc(stream), &mut rx).await; + }) + }); +} + +pub(crate) fn spawn_listener(server_id: i64, path: String, shutdown_rx: oneshot::Receiver<()>) { + perry_ffi::spawn_async(async move { + if let Err(error) = run_listener(server_id, path.clone(), shutdown_rx).await { + push_event(PendingNetEvent::ServerError( + server_id, + format!("bind {path}: {error}"), + )); + } + push_event(PendingNetEvent::ServerClose(server_id)); + if let Ok(mut servers) = statics::servers().lock() { + if let Some(server) = servers.get_mut(&server_id) { + server.listening = false; + } + } + }); +} + +#[cfg(unix)] +async fn connect_path(path: &str) -> io::Result> { + UnixStream::connect(path) + .await + .map(|stream| Box::new(stream) as Box) +} + +#[cfg(unix)] +async fn run_listener( + server_id: i64, + path: String, + mut shutdown_rx: oneshot::Receiver<()>, +) -> io::Result<()> { + let listener = UnixListener::bind(&path)?; + push_event(PendingNetEvent::ServerListening(server_id)); + + loop { + tokio::select! { + accepted = listener.accept() => match accepted { + Ok((stream, _)) => { + if let Some(info) = server_state::should_drop_ipc_connection(server_id) { + push_event(PendingNetEvent::ServerDrop(server_id, info)); + } else { + register_accepted_transport( + server_id, + Transport::Ipc(Box::new(stream)), + None, + ); + } + } + Err(error) => { + push_event(PendingNetEvent::ServerError( + server_id, + format!("accept: {error}"), + )); + } + }, + _ = &mut shutdown_rx => break, + } + } + + drop(listener); + // Tokio deliberately leaves filesystem socket nodes behind. Only unlink + // after our listener has closed; bind failures never remove someone else's + // endpoint. + match std::fs::remove_file(&path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +#[cfg(windows)] +async fn connect_path(path: &str) -> io::Result> { + loop { + match ClientOptions::new().open(path) { + Ok(stream) => { + return Ok(Box::new(stream) as Box); + } + // ERROR_PIPE_BUSY: all instances are serving clients. Match + // Node/libuv's wait-and-retry behavior rather than reporting a + // transient connector failure. + Err(error) if error.raw_os_error() == Some(231) => { + tokio::time::sleep(Duration::from_millis(10)).await; + } + Err(error) => return Err(error), + } + } +} + +#[cfg(windows)] +fn create_pipe_server(path: &str, first: bool) -> io::Result { + ServerOptions::new().first_pipe_instance(first).create(path) +} + +#[cfg(windows)] +async fn run_listener( + server_id: i64, + path: String, + mut shutdown_rx: oneshot::Receiver<()>, +) -> io::Result<()> { + let mut listener = create_pipe_server(&path, true)?; + push_event(PendingNetEvent::ServerListening(server_id)); + + loop { + tokio::select! { + connected = listener.connect() => { + connected?; + let stream = listener; + // A Windows named-pipe instance accepts exactly one client. + // Create the next instance before publishing the accepted one + // so concurrent connectors do not observe a needless gap. + listener = create_pipe_server(&path, false)?; + if let Some(info) = server_state::should_drop_ipc_connection(server_id) { + push_event(PendingNetEvent::ServerDrop(server_id, info)); + drop(stream); + } else { + register_accepted_transport( + server_id, + Transport::Ipc(Box::new(stream)), + None, + ); + } + } + _ = &mut shutdown_rx => break, + } + } + Ok(()) +} + +#[cfg(not(any(unix, windows)))] +async fn connect_path(_path: &str) -> io::Result> { + Err(io::Error::new( + io::ErrorKind::Unsupported, + "local IPC sockets are unsupported on this platform", + )) +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicU64, Ordering}; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + static NEXT_TEST_PIPE: AtomicU64 = AtomicU64::new(1); + + fn unique_name() -> String { + let suffix = NEXT_TEST_PIPE.fetch_add(1, Ordering::Relaxed); + #[cfg(windows)] + return format!(r"\\.\pipe\perry-ext-net-{}-{suffix}", std::process::id()); + #[cfg(unix)] + return std::env::temp_dir() + .join(format!( + "perry-ext-net-{}-{suffix}.sock", + std::process::id() + )) + .to_string_lossy() + .into_owned(); + #[cfg(not(any(unix, windows)))] + return String::new(); + } + + #[test] + fn nanboxed_string_is_recognized_as_an_ipc_path() { + let path = unique_name(); + let header = perry_ffi::alloc_string(&path).as_raw(); + let value = f64::from_bits(perry_ffi::nanbox_string_bits(header)); + assert_eq!(unsafe { super::string_value(value) }, Some(path)); + } + + #[cfg(windows)] + #[tokio::test] + async fn named_pipe_stream_round_trip() { + let path = unique_name(); + let mut server = super::create_pipe_server(&path, true).unwrap(); + let connect_path = path.clone(); + let client = tokio::spawn(async move { super::connect_path(&connect_path).await }); + server.connect().await.unwrap(); + let mut client = super::Transport::Ipc(client.await.unwrap().unwrap()); + + client.write_all(b"ping").await.unwrap(); + let mut request = [0; 4]; + server.read_exact(&mut request).await.unwrap(); + assert_eq!(&request, b"ping"); + + server.write_all(b"pong").await.unwrap(); + let mut response = [0; 4]; + client.read_exact(&mut response).await.unwrap(); + assert_eq!(&response, b"pong"); + } + + #[cfg(unix)] + #[tokio::test] + async fn unix_socket_stream_round_trip() { + let path = unique_name(); + let listener = super::UnixListener::bind(&path).unwrap(); + let connect_path = path.clone(); + let client = tokio::spawn(async move { super::connect_path(&connect_path).await }); + let (mut server, _) = listener.accept().await.unwrap(); + let mut client = super::Transport::Ipc(client.await.unwrap().unwrap()); + + client.write_all(b"ping").await.unwrap(); + let mut request = [0; 4]; + server.read_exact(&mut request).await.unwrap(); + assert_eq!(&request, b"ping"); + + server.write_all(b"pong").await.unwrap(); + let mut response = [0; 4]; + client.read_exact(&mut response).await.unwrap(); + assert_eq!(&response, b"pong"); + drop(listener); + std::fs::remove_file(path).unwrap(); + } +} + +#[cfg(not(any(unix, windows)))] +async fn run_listener( + _server_id: i64, + _path: String, + _shutdown_rx: oneshot::Receiver<()>, +) -> io::Result<()> { + Err(io::Error::new( + io::ErrorKind::Unsupported, + "local IPC sockets are unsupported on this platform", + )) +} diff --git a/crates/perry-ext-net/src/lib.rs b/crates/perry-ext-net/src/lib.rs index 73372b5626..54ae80a332 100644 --- a/crates/perry-ext-net/src/lib.rs +++ b/crates/perry-ext-net/src/lib.rs @@ -71,6 +71,7 @@ mod handle_ids; pub(crate) use handle_ids::{next_id, next_id_or_throw}; mod dispatch; mod dispatch_custody; +mod ipc; mod socket_emit; pub use socket_emit::{ js_ext_net_register_http_agent_socket_event_hook, js_ext_net_set_http_agent_phase, @@ -226,6 +227,9 @@ pub(crate) struct ServerState { pub shutdown_tx: Option>, pub bound_port: u16, pub bound_host: String, + /// Named-pipe / Unix-domain-socket path for an IPC listener. TCP servers + /// leave this unset and use `bound_host` + `bound_port`. + pub bound_path: Option, pub listening: bool, pub active_connections: usize, pub pending_connections: usize, @@ -489,7 +493,7 @@ where /// `net.createConnection(...)` / `net.connect(...)` — returns a handle /// immediately; connection happens in the background and emits -/// `'connect'` or `'error'`. Supports both Node overloads: +/// `'connect'` or `'error'`. Supports Node's TCP and IPC overloads: /// /// - Positional: `net.connect(port, host, cb?)`. `arg1_f64` is the /// port as a regular f64 number, `arg2_f64` carries the host as a @@ -499,6 +503,7 @@ where /// `port`; `arg2_f64` is the optional `connectListener`. In this /// form `arg3_f64` is unused (the dispatch table pads it with /// `undefined`). Issue #770. +/// - IPC: `net.connect(path, cb?)` or `net.connect({ path }, cb?)`. /// /// The `connectListener` (whichever slot it ends up in) is /// auto-registered as a `'connect'` listener on the new socket @@ -527,27 +532,21 @@ pub unsafe extern "C" fn js_ext_net_socket_connect( #[no_mangle] pub unsafe extern "C" fn js_net_socket_connect(arg1_f64: f64, arg2_f64: f64, arg3_f64: f64) -> i64 { - /// Register `cb_f64` as a `'connect'` listener on `handle` if it - /// carries a real closure pointer. No-op otherwise. - fn register_connect_cb(handle: i64, cb_f64: f64) { - if handle == 0 || !is_nanboxed_pointer(cb_f64) { - return; - } - let cb_ptr = unsafe { unbox_pointer(cb_f64) } as i64; - if cb_ptr == 0 { - return; - } - let mut listeners = statics::listeners().lock().unwrap(); - listeners - .entry(handle) - .or_default() - .entry("connect".to_string()) - .or_default() - .push(cb_ptr); + // Path overload: `net.connect(path[, cb])`. + if let Some(path) = ipc::string_value(arg1_f64) { + let handle = ipc::spawn_socket(path); + ipc::register_connect_cb(handle, arg2_f64); + return handle; } if is_nanboxed_pointer(arg1_f64) { - // Options-object overload: extract host/port from the object. + // Options-object overload. A `path` selects local IPC before the TCP + // host/port fields are considered, matching Node's normalization. + if let Some(path) = get_object_string_field(arg1_f64, "path") { + let handle = ipc::spawn_socket(path); + ipc::register_connect_cb(handle, arg2_f64); + return handle; + } let host = match get_object_string_field(arg1_f64, "host") .or_else(|| get_object_string_field(arg1_f64, "hostname")) { @@ -564,7 +563,7 @@ pub unsafe extern "C" fn js_net_socket_connect(arg1_f64: f64, arg2_f64: f64, arg }; let handle = spawn_socket_task(host, port, /* direct_tls: */ None); // connectListener lives in arg2 for the options form. - register_connect_cb(handle, arg2_f64); + ipc::register_connect_cb(handle, arg2_f64); return handle; } // Positional overload: arg1 is the port number, arg2 is the host @@ -588,7 +587,7 @@ pub unsafe extern "C" fn js_net_socket_connect(arg1_f64: f64, arg2_f64: f64, arg js_net_validate_connect_port(arg1_f64); let port = arg1_f64 as u16; let handle = spawn_socket_task(host, port, /* direct_tls: */ None); - register_connect_cb(handle, listener_f64); + ipc::register_connect_cb(handle, listener_f64); handle } @@ -656,6 +655,7 @@ pub unsafe extern "C" fn js_net_create_server( shutdown_tx: None, bound_port: 0, bound_host: String::new(), + bound_path: None, listening: false, active_connections: 0, pending_connections: 0, @@ -679,9 +679,9 @@ pub unsafe extern "C" fn js_net_create_server( // ─── FFI: net.Server.listen / .close / .address / .on ──────────────────────── -/// `server.listen(port, callback?)` — bind a tokio `TcpListener` on -/// `0.0.0.0:port` and spawn an accept loop on the shared multi-thread -/// runtime. The `callback` (a NaN-boxed closure pointer in the codegen's +/// `server.listen(port | path, callback?)` — bind TCP, a Windows named pipe, +/// or a Unix-domain socket and spawn an accept loop on the shared runtime. +/// The `callback` (a NaN-boxed closure pointer in the codegen's /// NA_PTR slot, raw i64 here after unboxing in lower_call.rs) is /// registered as a one-shot `'listening'` listener; when the bind /// resolves, the accept-loop task pushes a `ServerListening` event so @@ -704,13 +704,25 @@ pub unsafe extern "C" fn js_net_server_listen(handle: i64, port: f64, arg2: f64, 0 => js_net_callback_ptr(arg2), cb => cb, }; - // #2013: a numeric `port` must be an integer in [0, 65536); Node throws - // RangeError [ERR_SOCKET_BAD_PORT] otherwise. (A string is a pipe path and - // is left alone.) - js_net_validate_listen_port(port); - let port_u16 = port as u16; - let host = string_from_header_i64(js_get_string_pointer_unified(arg2)) - .unwrap_or_else(|| "0.0.0.0".to_string()); + let path = ipc::string_value(port) + .or_else(|| is_nanboxed_pointer(port).then(|| get_object_string_field(port, "path"))?); + let (port_u16, host) = if path.is_some() { + (0, String::new()) + } else if is_nanboxed_pointer(port) { + let option_port = get_object_number_field(port, "port").unwrap_or(0.0); + js_net_validate_listen_port(option_port); + let option_host = get_object_string_field(port, "host") + .filter(|host| !host.is_empty()) + .unwrap_or_else(|| "0.0.0.0".to_string()); + (option_port as u16, option_host) + } else { + // #2013: a numeric `port` must be an integer in [0, 65536); Node throws + // RangeError [ERR_SOCKET_BAD_PORT] otherwise. + js_net_validate_listen_port(port); + let host = string_from_header_i64(js_get_string_pointer_unified(arg2)) + .unwrap_or_else(|| "0.0.0.0".to_string()); + (port as u16, host) + }; let (shutdown_tx, mut shutdown_rx) = oneshot::channel::<()>(); @@ -728,6 +740,7 @@ pub unsafe extern "C" fn js_net_server_listen(handle: i64, port: f64, arg2: f64, s.shutdown_tx = Some(shutdown_tx); s.bound_port = port_u16; s.bound_host = host.clone(); + s.bound_path = path.clone(); s.listening = true; } @@ -748,6 +761,11 @@ pub unsafe extern "C" fn js_net_server_listen(handle: i64, port: f64, arg2: f64, let host_for_spawn = host.clone(); let server_id = handle; + if let Some(path) = path { + ipc::spawn_listener(server_id, path, shutdown_rx); + return; + } + // Run the accept loop cooperatively on Perry's shared multi-thread runtime // via `spawn_async` — no throwaway current-thread runtime, no blocking-pool // thread held for the server's life. The shared runtime owns the I/O @@ -810,34 +828,6 @@ pub unsafe extern "C" fn js_net_server_listen(handle: i64, port: f64, arg2: f64, push_event(PendingNetEvent::ServerDrop(server_id, info)); continue; } - // Allocate a fresh Socket handle that - // shares the existing socket machinery - // (run_socket_task, command channel, - // 'data'/'end'/'close'/'error' pump - // dispatch). The accept side doesn't - // need a tokio TcpStream::connect — we - // already have the stream — so we - // bypass `spawn_socket_task` (which - // calls TcpStream::connect inside) and - // call `run_socket_task` directly with - // the accepted stream. - let socket_id = next_id(); - // #6441: on a background thread there is no JS frame - // to unwind to, so exhaustion can't throw here. - // Drop the accepted stream — closing the connection, - // the EMFILE-style degradation Node applies when it - // can't accept — rather than register a phantom - // socket under the `0` sentinel. Refuse quietly: once - // the band is exhausted every accept fails, so an - // 'error' event per connection would flood a hot - // loop; the synchronous client-facing entry points - // still surface a throwable EMFILE. - if socket_id == perry_ffi::INVALID_HANDLE { - server_state::cancel_pending_connection(server_id); - drop(stream); - continue; - } - let (tx, rx) = mpsc::unbounded_channel::(); // Node sets TCP_NODELAY on every accepted socket by // default (Nagle off). Match that so small writes // aren't delayed waiting to coalesce; a later @@ -849,64 +839,11 @@ pub unsafe extern "C" fn js_net_server_listen(handle: i64, port: f64, arg2: f64, // bound port/family instead of returning // undefined. let accepted_local = stream.local_addr().ok(); - statics::sockets().lock().unwrap().insert( - socket_id, - SocketState { - cmd_tx: tx, - pending_rx: None, - is_open: true, - refed: true, - local_addr: accepted_local, - raw: None, - destroyed: false, - bytes_read: 0, - bytes_written: 0, - timeout: None, - type_of_service: 0, - server_id: Some(server_id), - server_connection_active: false, - tls: TlsSocketMetadata::default(), - }, + ipc::register_accepted_transport( + server_id, + Transport::Plain(stream), + accepted_local, ); - statics::listeners() - .lock() - .unwrap() - .insert(socket_id, HashMap::new()); - - // Surface the new socket to the user's - // `'connection'` listener on the main - // thread *before* spawning the read - // loop — the listener typically registers - // its own `.on('data', ...)` handlers - // and we want those in place before - // bytes start arriving. The accepted - // stream's read loop spawns next. - push_event(PendingNetEvent::ServerConnection( - server_id, socket_id, false, - )); - - // Spawn the per-socket read/write loop on - // the same shared runtime as this accept - // loop. A direct `tokio::spawn` (not - // `spawn_socket_runner`, which routes - // through the `perry_ffi::spawn_async` FFI - // shim) is correct here because we're - // already inside a task on the shared - // runtime — `tokio::spawn` lands on it - // directly, skipping the round-trip back - // out through C. The shim only matters at - // the FFI entry points that cross into - // Rust-from-C, where no ambient runtime - // task exists yet. - tokio::spawn(async move { - let mut rx = rx; - run_socket_task( - socket_id, - Transport::Plain(stream), - &mut rx, - ) - .await; - }); } Err(e) => { push_event(PendingNetEvent::ServerError( @@ -982,6 +919,12 @@ pub unsafe extern "C" fn js_net_server_address(handle: i64) -> *mut StringHeader let json = match statics::servers().lock() { Ok(g) => match g.get(&handle) { Some(s) if s.listening => { + if let Some(path) = &s.bound_path { + return alloc_string( + &serde_json::to_string(path).unwrap_or_else(|_| "null".to_string()), + ) + .as_raw(); + } let family = if s.bound_host.contains(':') { "IPv6" } else { @@ -1023,8 +966,8 @@ pub unsafe extern "C" fn js_net_server_on(handle: i64, event_ptr: i64, cb: i64) // ─── FFI: socket.connect(port, host) (instance method on existing handle) ───── -/// `socket.connect(port, host)` — initiates a TCP connection on a socket -/// previously allocated by `new net.Socket()`. Pulls its receiver out of +/// `socket.connect(port, host)` / `socket.connect(path)` — initiates a TCP or +/// IPC connection on a socket previously allocated by `new net.Socket()`. Pulls its receiver out of /// the `SocketState::pending_rx` slot rather than allocating a fresh /// channel, so any listener already registered (`sock.on('data', cb)`) /// sees the same handle id once the connect completes. @@ -1033,21 +976,63 @@ pub unsafe extern "C" fn js_net_server_on(handle: i64, event_ptr: i64, cb: i64) /// /// See `js_net_socket_connect`. #[no_mangle] -pub unsafe extern "C" fn js_net_socket_method_connect(handle: i64, port: f64, host_ptr: i64) { - // #2013: validate the port first (RangeError [ERR_SOCKET_BAD_PORT]), - // before any host handling, matching Node's `Socket.prototype.connect`. - js_net_validate_connect_port(port); - let host = match string_from_header_i64(host_ptr) { - Some(h) => h, - None => { - push_event(PendingNetEvent::Error( - handle, - "socket.connect: invalid host string".to_string(), - )); +pub unsafe extern "C" fn js_ext_net_socket_method_connect( + handle: i64, + arg1: f64, + arg2: f64, + arg3: f64, +) { + js_net_socket_method_connect(handle, arg1, arg2, arg3); +} + +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_method_connect( + handle: i64, + arg1: f64, + arg2: f64, + arg3: f64, +) { + if let Some(path) = ipc::string_value(arg1) { + ipc::register_connect_cb(handle, arg2); + ipc::connect_existing(handle, path); + return; + } + + let (host, port, callback) = if is_nanboxed_pointer(arg1) { + if let Some(path) = get_object_string_field(arg1, "path") { + ipc::register_connect_cb(handle, arg2); + ipc::connect_existing(handle, path); return; } + let port = match get_object_number_field(arg1, "port") { + Some(port) => port, + None => { + push_event(PendingNetEvent::Error( + handle, + "socket.connect: options.port or options.path is required".to_string(), + )); + return; + } + }; + let host = get_object_string_field(arg1, "host") + .or_else(|| get_object_string_field(arg1, "hostname")) + .filter(|host| !host.is_empty()) + .unwrap_or_else(|| "localhost".to_string()); + (host, port, arg2) + } else { + let host = ipc::string_value(arg2); + let callback = if host.is_some() { arg3 } else { arg2 }; + ( + host.unwrap_or_else(|| "127.0.0.1".to_string()), + arg1, + callback, + ) }; + // #2013: validate before truncating, matching Node's synchronous + // ERR_SOCKET_BAD_PORT behavior for positional and options overloads. + js_net_validate_connect_port(port); let port = port as u16; + ipc::register_connect_cb(handle, callback); let rx = { let mut guard = statics::sockets().lock().unwrap(); @@ -1374,6 +1359,12 @@ pub(crate) async fn run_socket_task( transport = Some(already_tls); let _ = reply.send(Err("socket is already TLS".to_string())); } + Some(ipc @ Transport::Ipc(_)) => { + transport = Some(ipc); + let _ = reply.send(Err( + "TLS upgrade is unsupported for IPC sockets".to_string(), + )); + } None => { let _ = reply.send(Err("socket closed".to_string())); break; diff --git a/crates/perry-ext-net/src/server_state.rs b/crates/perry-ext-net/src/server_state.rs index 5c03dcdbb3..9feadc9dc3 100644 --- a/crates/perry-ext-net/src/server_state.rs +++ b/crates/perry-ext-net/src/server_state.rs @@ -220,6 +220,18 @@ pub(crate) fn build_drop_object(info: &DropInfo) -> f64 { } pub(crate) fn should_drop_connection(server_id: i64, stream: &TcpStream) -> Option { + reserve_connection(server_id, stream.local_addr().ok(), stream.peer_addr().ok()) +} + +pub(crate) fn should_drop_ipc_connection(server_id: i64) -> Option { + reserve_connection(server_id, None, None) +} + +fn reserve_connection( + server_id: i64, + local: Option, + remote: Option, +) -> Option { let mut servers = statics::servers().lock().ok()?; let server = servers.get_mut(&server_id)?; if server @@ -227,10 +239,7 @@ pub(crate) fn should_drop_connection(server_id: i64, stream: &TcpStream) -> Opti .is_some_and(|max| server.active_connections + server.pending_connections >= max) && server.drop_max_connection.unwrap_or(false) { - return Some(DropInfo { - local: stream.local_addr().ok(), - remote: stream.peer_addr().ok(), - }); + return Some(DropInfo { local, remote }); } server.pending_connections += 1; None @@ -265,6 +274,26 @@ pub(crate) fn begin_local_connect(host: &str, port: u16) -> Option<(i64, bool)> Some((*server_id, expects_drop)) } +pub(crate) fn begin_local_path_connect(path: &str) -> Option<(i64, bool)> { + let mut servers = statics::servers().lock().ok()?; + let (server_id, server) = servers + .iter_mut() + .find(|(_, server)| server.listening && server.bound_path.as_deref() == Some(path))?; + let completed = connection_order_state() + .lock() + .unwrap() + .completed_local_connects + .get(server_id) + .copied() + .unwrap_or(0); + let expects_drop = server.drop_max_connection.unwrap_or(false) + && server.max_connections.is_some_and(|max| { + server.active_connections + server.pending_connections + completed >= max + }); + server.pending_local_connect_events += 1; + Some((*server_id, expects_drop)) +} + fn complete_local_connect(local_server: Option<(i64, bool)>, connected: bool) { let Some((server_id, expects_drop)) = local_server else { return; diff --git a/crates/perry-ext-net/src/test_async_shims.rs b/crates/perry-ext-net/src/test_async_shims.rs index 4a80fc1da0..fb51459b10 100644 --- a/crates/perry-ext-net/src/test_async_shims.rs +++ b/crates/perry-ext-net/src/test_async_shims.rs @@ -1,4 +1,4 @@ -use perry_ffi::Promise; +use perry_ffi::{NativeAsyncCompletion, Promise}; use std::ffi::c_void; // Unit-test binaries do not link the host stdlib/runtime archive that normally @@ -46,3 +46,70 @@ pub extern "C" fn perry_ffi_spawn_blocking_with_reactor( ) { invoke(ctx); } + +// The native-completion ABI is linked into perry-ffi even though ext-net's +// tests do not exercise it. Keep inert definitions here so the standalone +// crate test binary does not need the full Perry host archive. +#[no_mangle] +pub extern "C" fn perry_ffi_native_async_new(_flags: u32) -> *mut NativeAsyncCompletion { + std::ptr::null_mut() +} + +#[no_mangle] +pub extern "C" fn perry_ffi_native_async_promise( + _token: *mut NativeAsyncCompletion, +) -> *mut Promise { + std::ptr::null_mut() +} + +#[no_mangle] +pub extern "C" fn perry_ffi_native_async_resolve_bits( + _token: *mut NativeAsyncCompletion, + _bits: u64, +) -> i32 { + 0 +} + +#[no_mangle] +pub extern "C" fn perry_ffi_native_async_reject_bits( + _token: *mut NativeAsyncCompletion, + _bits: u64, +) -> i32 { + 0 +} + +#[no_mangle] +pub extern "C" fn perry_ffi_native_async_reject_string( + _token: *mut NativeAsyncCompletion, + _data: *const u8, + _len: usize, +) -> i32 { + 0 +} + +#[no_mangle] +pub extern "C" fn perry_ffi_native_async_cancel(_token: *mut NativeAsyncCompletion) -> i32 { + 0 +} + +#[no_mangle] +pub extern "C" fn perry_ffi_native_async_attach_handle( + _token: *mut NativeAsyncCompletion, + _handle_bits: u64, + _cleanup_flags: u32, +) -> i32 { + 0 +} + +#[no_mangle] +pub extern "C" fn perry_ffi_run_pending(_budget_ms: u64) {} + +#[no_mangle] +pub extern "C" fn js_tls_client_preflight( + _port: f64, + _servername_ptr: *const u8, + _servername_len: usize, + _options: f64, +) -> i32 { + 0 +} diff --git a/crates/perry-ext-net/src/transport.rs b/crates/perry-ext-net/src/transport.rs index 6d1897bf36..d725cb1abf 100644 --- a/crates/perry-ext-net/src/transport.rs +++ b/crates/perry-ext-net/src/transport.rs @@ -11,9 +11,14 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tokio::net::TcpStream; use tokio_rustls::client::TlsStream; +pub(crate) trait IpcStream: AsyncRead + AsyncWrite + Send + Unpin {} + +impl IpcStream for T where T: AsyncRead + AsyncWrite + Send + Unpin {} + pub(crate) enum Transport { Plain(TcpStream), Tls(Box>), + Ipc(Box), } impl Transport { @@ -25,6 +30,9 @@ impl Transport { match self { Transport::Plain(s) => s.set_nodelay(nodelay), Transport::Tls(s) => s.get_ref().0.set_nodelay(nodelay), + // Pipes do not use Nagle's algorithm. Node accepts setNoDelay on + // every net.Socket, including pipe-backed sockets, as a no-op. + Transport::Ipc(_) => Ok(()), } } @@ -35,6 +43,7 @@ impl Transport { match self { Transport::Plain(s) => s.nodelay(), Transport::Tls(s) => s.get_ref().0.nodelay(), + Transport::Ipc(_) => Ok(false), } } } @@ -48,6 +57,7 @@ impl AsyncRead for Transport { match self.get_mut() { Transport::Plain(s) => Pin::new(s).poll_read(cx, buf), Transport::Tls(s) => Pin::new(&mut **s).poll_read(cx, buf), + Transport::Ipc(s) => Pin::new(&mut **s).poll_read(cx, buf), } } } @@ -61,18 +71,21 @@ impl AsyncWrite for Transport { match self.get_mut() { Transport::Plain(s) => Pin::new(s).poll_write(cx, buf), Transport::Tls(s) => Pin::new(&mut **s).poll_write(cx, buf), + Transport::Ipc(s) => Pin::new(&mut **s).poll_write(cx, buf), } } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match self.get_mut() { Transport::Plain(s) => Pin::new(s).poll_flush(cx), Transport::Tls(s) => Pin::new(&mut **s).poll_flush(cx), + Transport::Ipc(s) => Pin::new(&mut **s).poll_flush(cx), } } fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match self.get_mut() { Transport::Plain(s) => Pin::new(s).poll_shutdown(cx), Transport::Tls(s) => Pin::new(&mut **s).poll_shutdown(cx), + Transport::Ipc(s) => Pin::new(&mut **s).poll_shutdown(cx), } } } diff --git a/crates/perry-stdlib/src/common/dispatch/fastify_net_zlib.rs b/crates/perry-stdlib/src/common/dispatch/fastify_net_zlib.rs index 7d942e59b2..3c9bd83f50 100644 --- a/crates/perry-stdlib/src/common/dispatch/fastify_net_zlib.rs +++ b/crates/perry-stdlib/src/common/dispatch/fastify_net_zlib.rs @@ -194,7 +194,7 @@ pub(crate) unsafe fn dispatch_external_net_socket(handle: i64, method: &str, arg // both-archives link — the registration is dropped and the socket's // 'data' events never reach JS. Use ext-net's distinct symbols. fn js_ext_net_socket_on(handle: i64, event_ptr: i64, cb_ptr: i64); - fn js_net_socket_method_connect(handle: i64, port: f64, host_ptr: i64); + fn js_ext_net_socket_method_connect(handle: i64, arg1: f64, arg2: f64, arg3: f64); fn js_net_socket_upgrade_tls( handle: i64, servername_ptr: i64, @@ -267,11 +267,12 @@ pub(crate) unsafe fn dispatch_external_net_socket(handle: i64, method: &str, arg js_ext_net_socket_on(handle, event_ptr, cb_ptr); nanbox_handle(handle) } - "connect" if args.len() >= 2 => { - let port = args[0]; - let host_ptr = unbox_to_i64(args[1]); - js_net_socket_method_connect(handle, port, host_ptr); - f64::from_bits(0x7FFC_0000_0000_0001) + "connect" if !args.is_empty() => { + let undefined = f64::from_bits(0x7FFC_0000_0000_0001); + let arg2 = args.get(1).copied().unwrap_or(undefined); + let arg3 = args.get(2).copied().unwrap_or(undefined); + js_ext_net_socket_method_connect(handle, args[0], arg2, arg3); + nanbox_handle(handle) } "upgradeToTLS" if !args.is_empty() => { let servername_ptr = unbox_to_i64(args[0]);