diff --git a/crates/perry-ext-http/src/server/https_server.rs b/crates/perry-ext-http/src/server/https_server.rs index ba538db684..5d58eae7f8 100644 --- a/crates/perry-ext-http/src/server/https_server.rs +++ b/crates/perry-ext-http/src/server/https_server.rs @@ -489,132 +489,133 @@ pub unsafe extern "C" fn js_node_https_server_listen(server_handle: i64, args_ar let request_tx_for_spawn = request_tx.clone(); let tls_config_for_spawn = tls_config; - perry_ffi::spawn_blocking_with_reactor(move || { - tokio::spawn(async move { - let listener = match TcpListener::from_std(std_listener) { - Ok(l) => l, - Err(e) => { - eprintln!("[node:https] tokio adopt failed: {}", e); - return; - } - }; - loop { - tokio::select! { - accepted = listener.accept() => { - match accepted { - Ok((stream, peer)) => { - // Node sets TCP_NODELAY on accepted connections by - // default. Honor the server's `noDelay` option - // (default true) on the raw TCP socket before the - // TLS handshake; the option persists through rustls. - crate::server::server::apply_accept_no_delay(&stream, no_delay); - let tls_config = tls_config_for_spawn.clone(); - let request_tx = request_tx_for_spawn.clone(); - // #4905/#4971 — register the connection so - // close()/closeAllConnections/ - // closeIdleConnections can reach this task - // from the main thread, and queue the - // 'connection' emit (Node fires it on the raw - // TCP connection, before the TLS handshake). - let conn_id = NEXT_CONNECTION_ID.fetch_add(1, Ordering::SeqCst); - let busy = Arc::new(AtomicUsize::new(0)); - let read_active = Arc::new(AtomicBool::new(false)); - let rewrite_chunked_header = Arc::new(AtomicBool::new(false)); - let close = Arc::new(tokio::sync::Notify::new()); - CONNECTIONS.lock().unwrap().insert( - conn_id, - TrackedConnection { - server_handle, - close: close.clone(), - busy: busy.clone(), - read_active: read_active.clone(), - }, - ); - if let Ok(mut q) = PENDING_CONNECTION_EVENTS.lock() { - q.push(server_handle); - } - tokio::spawn(async move { - let keylog = Arc::new(ConnectionKeyLog::default()); - let mut connection_config = (*tls_config).clone(); - connection_config.key_log = keylog.clone(); - let acceptor = TlsAcceptor::from(Arc::new(connection_config)); - let tls_stream = match acceptor.accept(stream).await { - Ok(s) => s, - Err(error) => { - queue_tls_client_error( - server_handle, - format!("TLS handshake failed: {error}"), - ); - CONNECTIONS.lock().unwrap().remove(&conn_id); - return; - } - }; - let negotiated_servername = tls_stream - .get_ref() - .1 - .server_name() - .map(String::from); - queue_tls_keylog(server_handle, keylog.drain()); - // Track read activity on the DECRYPTED - // stream — handshake bytes must not mark - // a request-less socket non-idle (#4971). - let io = TokioIo::new(ReadActivity::new( - tls_stream, - read_active.clone(), - rewrite_chunked_header.clone(), - )); - let close_for_service = close.clone(); - let service = service_fn(move |req: Request| { - let request_tx = request_tx.clone(); - let busy = busy.clone(); - let read_active = read_active.clone(); - let connection_close = close_for_service.clone(); - let rewrite_chunked_header = rewrite_chunked_header.clone(); - let negotiated_servername = negotiated_servername.clone(); - async move { - busy.fetch_add(1, Ordering::SeqCst); - read_active.store(false, Ordering::SeqCst); - let res = handle_https_request( - server_handle, - peer, - req, - request_tx, - connection_close, - rewrite_chunked_header, - negotiated_servername, - ) - .await; - busy.fetch_sub(1, Ordering::SeqCst); - res - } - }); - let mut builder = http1::Builder::new(); - builder.auto_date_header(false).title_case_headers(true); - let conn = builder.serve_connection(io, service).with_upgrades(); - tokio::pin!(conn); - tokio::select! { - result = &mut conn => { - // Common when the client closes - // mid-request — silenced. - let _ = result; - } - _ = close.notified() => { - // close()/closeAllConnections/ - // closeIdleConnections: dropping - // the pinned connection closes the - // socket immediately. - } + // Use the same explicit reactor-owned scheduling path as the plain HTTP + // listener. HTTPS supports the same attached WebSocket-server link shape, + // so it must not depend on an ambient Tokio context either (#8747). + perry_ffi::spawn_async(async move { + let listener = match TcpListener::from_std(std_listener) { + Ok(l) => l, + Err(e) => { + eprintln!("[node:https] tokio adopt failed: {}", e); + return; + } + }; + loop { + tokio::select! { + accepted = listener.accept() => { + match accepted { + Ok((stream, peer)) => { + // Node sets TCP_NODELAY on accepted connections by + // default. Honor the server's `noDelay` option + // (default true) on the raw TCP socket before the + // TLS handshake; the option persists through rustls. + crate::server::server::apply_accept_no_delay(&stream, no_delay); + let tls_config = tls_config_for_spawn.clone(); + let request_tx = request_tx_for_spawn.clone(); + // #4905/#4971 — register the connection so + // close()/closeAllConnections/ + // closeIdleConnections can reach this task + // from the main thread, and queue the + // 'connection' emit (Node fires it on the raw + // TCP connection, before the TLS handshake). + let conn_id = NEXT_CONNECTION_ID.fetch_add(1, Ordering::SeqCst); + let busy = Arc::new(AtomicUsize::new(0)); + let read_active = Arc::new(AtomicBool::new(false)); + let rewrite_chunked_header = Arc::new(AtomicBool::new(false)); + let close = Arc::new(tokio::sync::Notify::new()); + CONNECTIONS.lock().unwrap().insert( + conn_id, + TrackedConnection { + server_handle, + close: close.clone(), + busy: busy.clone(), + read_active: read_active.clone(), + }, + ); + if let Ok(mut q) = PENDING_CONNECTION_EVENTS.lock() { + q.push(server_handle); + } + tokio::spawn(async move { + let keylog = Arc::new(ConnectionKeyLog::default()); + let mut connection_config = (*tls_config).clone(); + connection_config.key_log = keylog.clone(); + let acceptor = TlsAcceptor::from(Arc::new(connection_config)); + let tls_stream = match acceptor.accept(stream).await { + Ok(s) => s, + Err(error) => { + queue_tls_client_error( + server_handle, + format!("TLS handshake failed: {error}"), + ); + CONNECTIONS.lock().unwrap().remove(&conn_id); + return; + } + }; + let negotiated_servername = tls_stream + .get_ref() + .1 + .server_name() + .map(String::from); + queue_tls_keylog(server_handle, keylog.drain()); + // Track read activity on the DECRYPTED + // stream — handshake bytes must not mark + // a request-less socket non-idle (#4971). + let io = TokioIo::new(ReadActivity::new( + tls_stream, + read_active.clone(), + rewrite_chunked_header.clone(), + )); + let close_for_service = close.clone(); + let service = service_fn(move |req: Request| { + let request_tx = request_tx.clone(); + let busy = busy.clone(); + let read_active = read_active.clone(); + let connection_close = close_for_service.clone(); + let rewrite_chunked_header = rewrite_chunked_header.clone(); + let negotiated_servername = negotiated_servername.clone(); + async move { + busy.fetch_add(1, Ordering::SeqCst); + read_active.store(false, Ordering::SeqCst); + let res = handle_https_request( + server_handle, + peer, + req, + request_tx, + connection_close, + rewrite_chunked_header, + negotiated_servername, + ) + .await; + busy.fetch_sub(1, Ordering::SeqCst); + res } - CONNECTIONS.lock().unwrap().remove(&conn_id); }); - } - Err(e) => eprintln!("[node:https] accept error: {}", e), + let mut builder = http1::Builder::new(); + builder.auto_date_header(false).title_case_headers(true); + let conn = builder.serve_connection(io, service).with_upgrades(); + tokio::pin!(conn); + tokio::select! { + result = &mut conn => { + // Common when the client closes + // mid-request — silenced. + let _ = result; + } + _ = close.notified() => { + // close()/closeAllConnections/ + // closeIdleConnections: dropping + // the pinned connection closes the + // socket immediately. + } + } + CONNECTIONS.lock().unwrap().remove(&conn_id); + }); } + Err(e) => eprintln!("[node:https] accept error: {}", e), } - _ = &mut shutdown_rx => break, } + _ = &mut shutdown_rx => break, } - }); + } }); // #4903 — queue the `'listening'` emit + the optional `cb` for the diff --git a/crates/perry-ext-http/src/server/server.rs b/crates/perry-ext-http/src/server/server.rs index bea8c7e00f..0aa207bd5f 100644 --- a/crates/perry-ext-http/src/server/server.rs +++ b/crates/perry-ext-http/src/server/server.rs @@ -716,38 +716,39 @@ fn spawn_rr_inject_loop( } }); - perry_ffi::spawn_blocking_with_reactor(move || { - tokio::spawn(async move { - loop { - tokio::select! { - maybe_fd = fd_rx.recv() => { - let Some(fd) = maybe_fd else { break }; - let std_stream = unsafe { std::net::TcpStream::from_raw_fd(fd) }; - if std_stream.set_nonblocking(true).is_err() { - continue; - } - let peer = std_stream - .peer_addr() - .unwrap_or_else(|_| SocketAddr::from(([0, 0, 0, 0], 0))); - match tokio::net::TcpStream::from_std(std_stream) { - Ok(stream) => { - // Match Node's per-connection `noDelay` (default on). - let _ = stream.set_nodelay(no_delay); - serve_http_connection( - server_handle, - stream, - peer, - request_tx_for_spawn.clone(), - upgrade_tx_for_spawn.clone(), - ); - } - Err(e) => eprintln!("[node:http] rr adopt failed: {}", e), + // Cross the FFI boundary as a future so Perry schedules the loop directly + // on its reactor-owned runtime. Relying on an ambient Tokio handle inside + // `spawn_blocking_with_reactor` is link-shape-sensitive (#8747). + perry_ffi::spawn_async(async move { + loop { + tokio::select! { + maybe_fd = fd_rx.recv() => { + let Some(fd) = maybe_fd else { break }; + let std_stream = unsafe { std::net::TcpStream::from_raw_fd(fd) }; + if std_stream.set_nonblocking(true).is_err() { + continue; + } + let peer = std_stream + .peer_addr() + .unwrap_or_else(|_| SocketAddr::from(([0, 0, 0, 0], 0))); + match tokio::net::TcpStream::from_std(std_stream) { + Ok(stream) => { + // Match Node's per-connection `noDelay` (default on). + let _ = stream.set_nodelay(no_delay); + serve_http_connection( + server_handle, + stream, + peer, + request_tx_for_spawn.clone(), + upgrade_tx_for_spawn.clone(), + ); } + Err(e) => eprintln!("[node:http] rr adopt failed: {}", e), } - _ = &mut shutdown_rx => break, } + _ = &mut shutdown_rx => break, } - }); + } }); } @@ -887,46 +888,44 @@ pub unsafe extern "C" fn js_node_http_server_listen(server_handle: i64, args_arr // whole listener lifetime in a GC-unsafe zone would disable `gc()` for // long-running servers without adding safety. - // The closure passed to `spawn_blocking_with_reactor` runs INSIDE - // a tokio worker task (perry-stdlib's shim wraps it in - // `runtime().spawn(async { invoke(...) })`), so calling - // `Handle::current().block_on(fut)` would panic with - // "Cannot start a runtime from within a runtime". Spawn the - // accept loop as a separate async task on the existing runtime - // and let the closure return immediately. - perry_ffi::spawn_blocking_with_reactor(move || { - tokio::spawn(async move { - let listener = match TcpListener::from_std(std_listener) { - Ok(l) => l, - Err(e) => { - eprintln!("[node:http] tokio adopt failed: {}", e); - return; - } - }; - loop { - tokio::select! { - accepted = listener.accept() => { - match accepted { - Ok((stream, peer)) => { - // Match Node's per-connection `noDelay` (default on). - let _ = stream.set_nodelay(no_delay); - serve_http_connection( - server_handle, - stream, - peer, - request_tx_for_spawn.clone(), - upgrade_tx_for_spawn.clone(), - ); - } - Err(e) => eprintln!("[node:http] accept error: {}", e), + // Schedule the accept future through Perry's explicit async bridge. + // `spawn_blocking_with_reactor(|| tokio::spawn(...))` depended on an + // ambient Tokio context inside the FFI callback; when `ws` changed the + // native-wrapper link shape, the spawned task reached `from_std` + // without the HTTP wrapper seeing a reactor and aborted (#8747). + // The server's `listening && refed` active gate above keeps Perry's + // event loop driving this detached future until `close()` fires. + perry_ffi::spawn_async(async move { + let listener = match TcpListener::from_std(std_listener) { + Ok(l) => l, + Err(e) => { + eprintln!("[node:http] tokio adopt failed: {}", e); + return; + } + }; + loop { + tokio::select! { + accepted = listener.accept() => { + match accepted { + Ok((stream, peer)) => { + // Match Node's per-connection `noDelay` (default on). + let _ = stream.set_nodelay(no_delay); + serve_http_connection( + server_handle, + stream, + peer, + request_tx_for_spawn.clone(), + upgrade_tx_for_spawn.clone(), + ); } + Err(e) => eprintln!("[node:http] accept error: {}", e), } - _ = &mut shutdown_rx => { - break; - } + } + _ = &mut shutdown_rx => { + break; } } - }); + } }); } diff --git a/crates/perry/tests/issue_8747_http_ws_shared_server.rs b/crates/perry/tests/issue_8747_http_ws_shared_server.rs new file mode 100644 index 0000000000..89917aeba9 --- /dev/null +++ b/crates/perry/tests/issue_8747_http_ws_shared_server.rs @@ -0,0 +1,115 @@ +//! Regression test for #8747: attaching `ws.WebSocketServer` to a +//! `node:http` server made the first ordinary HTTP request abort in +//! `TcpListener::from_std` with "there is no reactor running". +//! +//! The `ws` import changes the native-wrapper link shape. The HTTP accept loop +//! must therefore be scheduled through Perry's explicit shared-runtime bridge, +//! rather than relying on an ambient Tokio context in an FFI callback. + +use std::path::PathBuf; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +fn compile(dir: &std::path::Path, source: &str) -> PathBuf { + 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) + ); + output +} + +fn run_with_timeout(bin: &std::path::Path, secs: u64) -> (String, String) { + use std::io::Read; + + let mut child = Command::new(bin) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn compiled binary"); + let mut stdout_pipe = child.stdout.take().expect("piped stdout"); + let mut stderr_pipe = child.stderr.take().expect("piped stderr"); + let stdout_reader = std::thread::spawn(move || { + let mut output = String::new(); + let _ = stdout_pipe.read_to_string(&mut output); + output + }); + let stderr_reader = std::thread::spawn(move || { + let mut output = String::new(); + let _ = stderr_pipe.read_to_string(&mut output); + output + }); + + let deadline = Instant::now() + Duration::from_secs(secs); + loop { + match child.try_wait().expect("try_wait") { + Some(status) => { + let stdout = stdout_reader.join().unwrap_or_default(); + let stderr = stderr_reader.join().unwrap_or_default(); + assert!( + status.success(), + "#8747 regression: shared HTTP/WS server aborted\n\ + status: {status:?}\nstdout:\n{stdout}\nstderr:\n{stderr}" + ); + return (stdout, stderr); + } + None if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + let stdout = stdout_reader.join().unwrap_or_default(); + let stderr = stderr_reader.join().unwrap_or_default(); + panic!( + "#8747 regression: shared HTTP/WS server hung for >{secs}s\n\ + stdout:\n{stdout}\nstderr:\n{stderr}" + ); + } + None => std::thread::sleep(Duration::from_millis(50)), + } + } +} + +#[test] +fn plain_http_request_survives_attached_websocket_server() { + let dir = tempfile::tempdir().expect("tempdir"); + let bin = compile( + dir.path(), + r#" +import { createServer } from "node:http"; +import { WebSocketServer } from "ws"; + +const httpServer = createServer((_req, res) => { + res.writeHead(200); + res.end("http-ok"); +}); +const wss = new WebSocketServer({ server: httpServer }); + +await new Promise((resolve) => httpServer.listen(0, resolve)); +const port = (httpServer.address() as { port: number }).port; +const response = await fetch("http://127.0.0.1:" + port + "/"); +console.log(response.status, await response.text()); + +wss.close(); +httpServer.close(); +process.exit(0); +"#, + ); + let (stdout, stderr) = run_with_timeout(&bin, 30); + assert_eq!(stdout, "200 http-ok\n", "unexpected stderr:\n{stderr}"); +}