From ac79b7d3473053e5d63124229da0a4d096ea6171 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Mon, 24 Aug 2026 20:26:11 +0200 Subject: [PATCH] fix: mysql2 request isolation, forwarded-array loops, parent prototypes, reactor HTTP scheduling Lands #8765, #8767, #8768 and #8769. #8765 stops mysql2 prepared statements and pool transactions leaking state across requests: each SQL string and parameter vector lives in one owned request, a parameterless `query()` uses the text protocol, prepared statements are request-scoped, and registry-backed mutable connection references become serialized owned handles with safe close/release around in-flight work. #8767 admits arrays reached through one validated forwarding edge into version-stable indexed loops, canonicalizing the compiler-private local to the live array after the full header/fingerprint check. Per-iteration fingerprint guards are retained, so callback-driven growth or a GC still side-exits before the next effect, and invalid targets or longer chains fail closed to the generic loop. #8768 materializes ordinary parent prototypes. #8769 schedules HTTP and HTTPS accept loops through the reactor-owned async bridge, using the same path for Unix round-robin fd injection. Changelog fragments added for #8765, #8767 and #8769; none carried one or a skip-changelog label. No version bump. --- Cargo.lock | 1 + .../8765-mysql2-request-scoped-statements.md | 1 + .../8767-forwarded-arrays-versioned-loops.md | 1 + .../8768-runtime-parent-function-prototype.md | 4 + .../8769-http-ws-reactor-scheduling.md | 1 + .../src/codegen/index_method_clone_tests.rs | 53 ++ .../src/stmt/versioned_indexed_loop.rs | 56 +- .../perry-ext-http/src/server/https_server.rs | 241 ++++----- crates/perry-ext-http/src/server/server.rs | 127 +++-- crates/perry-ext-mysql2/Cargo.toml | 3 + crates/perry-ext-mysql2/src/lib.rs | 509 +++++++++++++----- .../perry-ext-mysql2/src/test_async_shims.rs | 103 ++++ .../src/object/class_registry/state.rs | 30 +- .../tests/issue_8747_http_ws_shared_server.rs | 115 ++++ .../versioned_indexed_loop_forwarding.rs | 108 ++++ ...ue_8745_8746_mysql2_operation_isolation.ts | 69 +++ 16 files changed, 1061 insertions(+), 361 deletions(-) create mode 100644 changelog.d/8765-mysql2-request-scoped-statements.md create mode 100644 changelog.d/8767-forwarded-arrays-versioned-loops.md create mode 100644 changelog.d/8768-runtime-parent-function-prototype.md create mode 100644 changelog.d/8769-http-ws-reactor-scheduling.md create mode 100644 crates/perry-ext-mysql2/src/test_async_shims.rs create mode 100644 crates/perry/tests/issue_8747_http_ws_shared_server.rs create mode 100644 crates/perry/tests/versioned_indexed_loop_forwarding.rs create mode 100644 test-files/test_issue_8745_8746_mysql2_operation_isolation.ts diff --git a/Cargo.lock b/Cargo.lock index 4645cbc6ac..2a9e494b60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6077,6 +6077,7 @@ version = "0.5.1519" dependencies = [ "chrono", "perry-ffi", + "perry-runtime", "sqlx", "tokio", ] diff --git a/changelog.d/8765-mysql2-request-scoped-statements.md b/changelog.d/8765-mysql2-request-scoped-statements.md new file mode 100644 index 0000000000..c4c32fc4b9 --- /dev/null +++ b/changelog.d/8765-mysql2-request-scoped-statements.md @@ -0,0 +1 @@ +Fixed mysql2 prepared statements and pool transactions leaking state across requests. Each SQL string and parameter vector now lives in one owned request, a parameterless `query()` uses the text protocol, and prepared statements are request-scoped so cached metadata cannot cross requests. Registry-backed mutable connection references are replaced with serialized owned handles, with safe close/release behaviour around in-flight work, and checked-out pool connections route through `query`, `execute`, `beginTransaction`, `commit`, `rollback` and `release`. diff --git a/changelog.d/8767-forwarded-arrays-versioned-loops.md b/changelog.d/8767-forwarded-arrays-versioned-loops.md new file mode 100644 index 0000000000..0ffc29b1cd --- /dev/null +++ b/changelog.d/8767-forwarded-arrays-versioned-loops.md @@ -0,0 +1 @@ +Version-stable indexed loops now admit arrays reached through one validated forwarding edge, canonicalizing the compiler-private local to the live array after the full header/fingerprint check. Per-iteration fingerprint guards are retained, so callback-driven growth or a GC still side-exits before the next effect, and invalid targets or longer forwarding chains fail closed to the generic loop. diff --git a/changelog.d/8768-runtime-parent-function-prototype.md b/changelog.d/8768-runtime-parent-function-prototype.md new file mode 100644 index 0000000000..6938e344b5 --- /dev/null +++ b/changelog.d/8768-runtime-parent-function-prototype.md @@ -0,0 +1,4 @@ +Fixed subclasses of runtime-valued ordinary functions failing during prototype +materialization when the parent function's lazy `.prototype` had not already +been read. `class Child extends Parent` now observes and links the same +prototype object as an ordinary `Parent.prototype` read. diff --git a/changelog.d/8769-http-ws-reactor-scheduling.md b/changelog.d/8769-http-ws-reactor-scheduling.md new file mode 100644 index 0000000000..f178988cec --- /dev/null +++ b/changelog.d/8769-http-ws-reactor-scheduling.md @@ -0,0 +1 @@ +Fixed shared HTTP/WS servers by scheduling HTTP and HTTPS accept loops through Perry's explicit reactor-owned async bridge, using the same path for Unix round-robin file-descriptor injection. Adds an end-to-end regression covering a `WebSocketServer` attached to a `node:http` server followed by a plain `fetch`. diff --git a/crates/perry-codegen/src/codegen/index_method_clone_tests.rs b/crates/perry-codegen/src/codegen/index_method_clone_tests.rs index 4138a603d6..452a3da5fa 100644 --- a/crates/perry-codegen/src/codegen/index_method_clone_tests.rs +++ b/crates/perry-codegen/src/codegen/index_method_clone_tests.rs @@ -515,6 +515,59 @@ fn checked_reader_callback_loop_versions_to_fast_and_resumable_slow_bodies() { ); } +#[test] +fn versioned_checked_reader_admission_canonicalizes_one_forwarding_edge() { + let ir = emit_versioned_checked_reader_loop(); + let iterate = function_body( + &ir, + "@perry_method_versioned_checked_reader_loop_ts__Reader__iterate(", + ); + let source_guard = iterate + .split("\nversioned_index.array.source_deref.") + .nth(1) + .and_then(|body| body.split("\nversioned_index.array.live_deref.").next()) + .unwrap_or_else(|| panic!("loop has no forwarding-source guard:\n{iterate}")); + let live_handle = source_guard + .lines() + .find(|line| line.contains(" = select i1") && line.contains(", i64 ")) + .and_then(|line| line.trim().split_once(" = ").map(|(name, _)| name)) + .unwrap_or_else(|| panic!("source guard has no selected live handle:\n{source_guard}")); + assert!( + source_guard.contains("and i8") + && source_guard.contains(", 128") + && source_guard.contains("load i64") + && source_guard.contains("label %versioned_index.array.live_deref.") + && source_guard.contains("label %versioned_index.loop.slow.preheader") + && !source_guard.contains(&format!("sub i64 {live_handle}, 8")), + "admission must select one forwarding target and validate its address before \ + reading its header:\n{source_guard}" + ); + let live_guard = iterate + .split("\nversioned_index.array.live_deref.") + .nth(1) + .and_then(|body| body.split("\nversioned_index.array.canonicalize.").next()) + .unwrap_or_else(|| panic!("loop has no selected-target header guard:\n{iterate}")); + assert!( + live_guard.contains(&format!("sub i64 {live_handle}, 8")) + && live_guard.contains("label %versioned_index.array.canonicalize.") + && live_guard.contains("label %versioned_index.loop.slow.preheader"), + "the selected target must be fully re-branded before admission:\n{live_guard}" + ); + let canonicalize = iterate + .split("\nversioned_index.array.canonicalize.") + .nth(1) + .and_then(|body| body.split("\nversioned_index.array.source_deref.").next()) + .unwrap_or_else(|| panic!("loop has no canonicalization block:\n{iterate}")); + assert!( + canonicalize.contains(&format!( + "or i64 {live_handle}, {}", + crate::nanbox::POINTER_TAG_I64 + )) && canonicalize.contains("store ptr addrspace(1)"), + "the uncaptured array local must be rewritten to the admitted live target so \ + iteration guards do not revisit an identity stub:\n{canonicalize}" + ); +} + #[test] fn guarded_read_can_follow_one_forwarding_edge_but_rechecks_the_live_header() { let ir = emit(); diff --git a/crates/perry-codegen/src/stmt/versioned_indexed_loop.rs b/crates/perry-codegen/src/stmt/versioned_indexed_loop.rs index 4332a291ab..f8079d7cf0 100644 --- a/crates/perry-codegen/src/stmt/versioned_indexed_loop.rs +++ b/crates/perry-codegen/src/stmt/versioned_indexed_loop.rs @@ -244,8 +244,12 @@ fn emit_array_admission( slow_label: &str, ) -> Option<(String, String)> { let local_slot = ctx.locals.get(&local_id)?.clone(); - let deref_idx = ctx.new_block("versioned_index.array.deref"); - let deref_label = ctx.block_label(deref_idx); + let source_deref_idx = ctx.new_block("versioned_index.array.source_deref"); + let source_deref_label = ctx.block_label(source_deref_idx); + let live_deref_idx = ctx.new_block("versioned_index.array.live_deref"); + let live_deref_label = ctx.block_label(live_deref_idx); + let canonicalize_idx = ctx.new_block("versioned_index.array.canonicalize"); + let canonicalize_label = ctx.block_label(canonicalize_idx); let heap_floor = crate::target_layout::heap_addr_lower_bound_inclusive(ctx.target_triple).to_string(); let heap_ceiling = @@ -262,10 +266,40 @@ fn emit_array_admission( let below_ceiling = ctx.block().icmp_ult(I64, &array_handle, &heap_ceiling); let in_heap = ctx.block().and(I1, &above_floor, &below_ceiling); let safe = ctx.block().and(I1, &is_pointer, &in_heap); - ctx.block().cond_br(&safe, &deref_label, slow_label); + ctx.block().cond_br(&safe, &source_deref_label, slow_label); - ctx.current_block = deref_idx; - let fingerprint_addr = ctx.block().sub(I64, &array_handle, "8"); + // Array growth leaves a forwarding stub at the identity-bearing address. + // Mirror the ordinary indexed-read guard: follow at most one edge, then + // validate the selected address before touching its header. A longer chain + // remains fail-closed and resumes the generic loop. + ctx.current_block = source_deref_idx; + let source_gc_type_addr = ctx.block().sub(I64, &array_handle, "8"); + let source_gc_type_ptr = ctx.block().inttoptr(I64, &source_gc_type_addr); + let source_gc_type = ctx.block().load(I8, &source_gc_type_ptr); + let source_is_array = ctx.block().icmp_eq(I8, &source_gc_type, "1"); + let source_flags_addr = ctx.block().sub(I64, &array_handle, "7"); + let source_flags_ptr = ctx.block().inttoptr(I64, &source_flags_addr); + let source_flags = ctx.block().load(I8, &source_flags_ptr); + let source_forwarded_bits = ctx.block().and(I8, &source_flags, "128"); + let source_is_forwarded = ctx.block().icmp_ne(I8, &source_forwarded_bits, "0"); + let source_ptr = ctx.block().inttoptr(I64, &array_handle); + let forwarding_target = ctx.block().load(I64, &source_ptr); + let follow_forwarding = ctx.block().and(I1, &source_is_array, &source_is_forwarded); + let live_handle = ctx.block().select( + I1, + &follow_forwarding, + I64, + &forwarding_target, + &array_handle, + ); + let live_above_floor = ctx.block().icmp_uge(I64, &live_handle, &heap_floor); + let live_below_ceiling = ctx.block().icmp_ult(I64, &live_handle, &heap_ceiling); + let live_in_heap = ctx.block().and(I1, &live_above_floor, &live_below_ceiling); + ctx.block() + .cond_br(&live_in_heap, &live_deref_label, slow_label); + + ctx.current_block = live_deref_idx; + let fingerprint_addr = ctx.block().sub(I64, &live_handle, "8"); let fingerprint_ptr = ctx.block().inttoptr(I64, &fingerprint_addr); let fingerprint = ctx.block().load_aligned(I128, &fingerprint_ptr, 8); let gc_header = ctx.block().trunc(I128, &fingerprint, I64); @@ -298,7 +332,17 @@ fn emit_array_admission( pass = ctx.block().and(I1, &pass, &length_sane); pass = ctx.block().and(I1, &pass, &capacity_sane); pass = ctx.block().and(I1, &pass, &length_within_capacity); - ctx.block().cond_br(&pass, success_label, slow_label); + ctx.block().cond_br(&pass, &canonicalize_label, slow_label); + + // Candidate analysis excludes rebinding and closure capture of this local, + // so replacing its internal root with the live address is unobservable. + // It also makes the existing per-iteration fingerprint guard O(1): a later + // growth/GC move turns this live address into a stub and side-exits before + // any effect, instead of re-walking an already-stale identity stub forever. + ctx.current_block = canonicalize_idx; + let live_box = crate::expr::nanbox_pointer_inline(ctx.block(), &live_handle); + ctx.block().store(DOUBLE, &live_box, &local_slot); + ctx.block().br(success_label); Some((local_slot, fingerprint)) } 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-ext-mysql2/Cargo.toml b/crates/perry-ext-mysql2/Cargo.toml index 5dc4a3be90..dcd6219727 100644 --- a/crates/perry-ext-mysql2/Cargo.toml +++ b/crates/perry-ext-mysql2/Cargo.toml @@ -24,3 +24,6 @@ chrono.workspace = true [dev-dependencies] perry-ffi = { workspace = true, features = ["runtime-link"] } +# Standalone extension tests need the runtime half of the test-only async FFI +# shims; production code still depends on perry-ffi only. +perry-runtime = { workspace = true, features = ["default", "stdlib"] } diff --git a/crates/perry-ext-mysql2/src/lib.rs b/crates/perry-ext-mysql2/src/lib.rs index 960ad26621..7ee3fe054b 100644 --- a/crates/perry-ext-mysql2/src/lib.rs +++ b/crates/perry-ext-mysql2/src/lib.rs @@ -15,15 +15,20 @@ //! adapter; followup once a wrapper actually demands it). use perry_ffi::{ - alloc_string, build_object_shape, get_handle_mut, js_array_alloc, js_array_get, js_array_push, + alloc_string, build_object_shape, js_array_alloc, js_array_get, js_array_push, js_object_alloc_with_shape, js_object_get_field, js_object_set_field, register_handle, - spawn_blocking, take_handle, ArrayHeader, Handle, JsPromise, JsValue, ObjectHeader, Promise, - StringHeader, + spawn_blocking, take_handle, with_handle, ArrayHeader, Handle, JsPromise, JsValue, + ObjectHeader, Promise, StringHeader, }; use sqlx::mysql::{MySqlConnection, MySqlPool, MySqlPoolOptions, MySqlRow}; use sqlx::pool::PoolConnection; use sqlx::{Column, Connection, MySql, Row, TypeInfo}; +use std::sync::Arc; use std::time::Duration; +use tokio::sync::Mutex; + +#[cfg(test)] +mod test_async_shims; const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 10; const DEFAULT_QUERY_TIMEOUT_SECS: u64 = 30; @@ -470,7 +475,7 @@ fn is_row_returning_query(sql: &str) -> bool { || upper.starts_with("WITH") } -#[derive(Clone, Debug)] +#[derive(Clone, Debug, PartialEq)] enum ParamValue { Null, String(String), @@ -479,6 +484,44 @@ enum ParamValue { Bool(bool), } +/// Everything needed to execute one mysql2 call, copied off the Perry heap +/// before the asynchronous work is scheduled. Keeping the SQL and its bind +/// values in one owned object makes it impossible for a later call to replace +/// either half while this request is waiting for a pool connection. +#[derive(Clone, Debug, PartialEq)] +struct QueryRequest { + sql: String, + params: Vec, + rows_as_array: bool, + /// `mysql2.query()` uses the text protocol when it has no values, whereas + /// `execute()` always represents a prepared statement. + force_prepared: bool, +} + +impl QueryRequest { + fn new( + sql: String, + params: Vec, + rows_as_array: bool, + force_prepared: bool, + ) -> Self { + Self { + sql, + params, + rows_as_array, + force_prepared, + } + } + + fn is_row_returning(&self) -> bool { + is_row_returning_query(&self.sql) + } + + fn uses_prepared_statement(&self) -> bool { + self.force_prepared || !self.params.is_empty() + } +} + unsafe fn extract_params_from_jsvalue(params: JsValue) -> Vec { let arr_ptr = params.as_pointer::(); if arr_ptr.is_null() { @@ -527,13 +570,135 @@ unsafe fn read_sql(sql_ptr: *const u8) -> String { // ── Connection ──────────────────────────────────────────────────── pub struct MysqlConnectionHandle { - pub connection: Option, + pub connection: Arc>>, } impl MysqlConnectionHandle { pub fn new(conn: MySqlConnection) -> Self { Self { - connection: Some(conn), + connection: Arc::new(Mutex::new(Some(conn))), + } + } +} + +#[derive(Clone)] +enum MysqlConnectionTarget { + Direct(Arc>>), + Pool(Arc>>>), +} + +/// Resolve either mysql2 connection handle family without returning a +/// registry-backed `'static` reference. The old `get_handle_mut` calls dropped +/// DashMap's guard before async work began, so overlapping workers could hold +/// aliased mutable references to the same connection wrapper. +fn connection_target(handle: Handle) -> Option { + with_handle::(handle, |wrapper| { + MysqlConnectionTarget::Direct(Arc::clone(&wrapper.connection)) + }) + .or_else(|| { + with_handle::(handle, |wrapper| { + MysqlConnectionTarget::Pool(Arc::clone(&wrapper.connection)) + }) + }) +} + +async fn execute_query_on_connection( + conn: &mut MySqlConnection, + request: &QueryRequest, +) -> Result { + let is_select = request.is_row_returning(); + + if !request.uses_prepared_statement() { + // SQLx's `query()` prepares even when there are no bind values. That + // needlessly put mysql2 `query("DROP ...")` / `query("CREATE ...")` + // calls into the per-connection statement cache beside parameterized + // `execute()` calls. Use MySQL's text protocol for the no-param query + // shape, matching mysql2 and keeping those statements out of the cache. + let raw = sqlx::raw_sql(sqlx::AssertSqlSafe(request.sql.clone())); + if is_select { + let rows = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + raw.fetch_all(conn), + ) + .await + .map_err(|_| "Query timed out".to_string())? + .map_err(|e| format!("Query failed: {}", e))?; + return Ok(QueryOutcome::Rows(raws_from_mysql_rows(rows))); + } + + let res = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + raw.execute(conn), + ) + .await + .map_err(|_| "Query timed out".to_string())? + .map_err(|e| format!("Query failed: {}", e))?; + return Ok(QueryOutcome::Executed { + affected_rows: res.rows_affected(), + last_insert_id: res.last_insert_id(), + }); + } + + // Build the SQLx query and all of its arguments from the same owned + // request immediately before execution. Nothing is shared with another + // mysql2 call, even while this future is waiting on I/O. + // Keep the prepared statement scoped to this request. SQLx's connection + // cache is where #8745 observed metadata from a neighboring statement + // being paired with this request's arguments; an ephemeral statement + // preserves mysql2 execute semantics without reusing that association. + let mut query = sqlx::query(sqlx::AssertSqlSafe(request.sql.clone())).persistent(false); + for param in &request.params { + query = match param { + ParamValue::Null => query.bind(Option::::None), + ParamValue::String(s) => query.bind(s.clone()), + ParamValue::Number(n) => query.bind(*n), + ParamValue::Int(i) => query.bind(*i), + ParamValue::Bool(b) => query.bind(*b), + }; + } + + if is_select { + let rows = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + query.fetch_all(conn), + ) + .await + .map_err(|_| "Query timed out".to_string())? + .map_err(|e| format!("Query failed: {}", e))?; + Ok(QueryOutcome::Rows(raws_from_mysql_rows(rows))) + } else { + let res = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + query.execute(conn), + ) + .await + .map_err(|_| "Query timed out".to_string())? + .map_err(|e| format!("Query failed: {}", e))?; + Ok(QueryOutcome::Executed { + affected_rows: res.rows_affected(), + last_insert_id: res.last_insert_id(), + }) + } +} + +async fn execute_query_on_target( + target: MysqlConnectionTarget, + request: &QueryRequest, +) -> Result { + match target { + MysqlConnectionTarget::Direct(connection) => { + let mut slot = connection.lock().await; + let conn = slot + .as_mut() + .ok_or_else(|| "Connection already closed".to_string())?; + execute_query_on_connection(conn, request).await + } + MysqlConnectionTarget::Pool(connection) => { + let mut slot = connection.lock().await; + let conn = slot + .as_mut() + .ok_or_else(|| "Pool connection released".to_string())?; + execute_query_on_connection(conn, request).await } } } @@ -575,8 +740,11 @@ pub extern "C" fn js_mysql2_connection_end(conn_handle: Handle) -> *mut Promise let promise = JsPromise::new(); let raw = promise.as_raw(); spawn_blocking(move || { - if let Some(mut wrapper) = take_handle::(conn_handle) { - if let Some(conn) = wrapper.connection.take() { + if let Some(wrapper) = take_handle::(conn_handle) { + let connection = Arc::clone(&wrapper.connection); + let conn = tokio::runtime::Handle::current() + .block_on(async move { connection.lock().await.take() }); + if let Some(conn) = conn { let result = tokio::runtime::Handle::current().block_on(conn.close()); match result { Ok(()) => promise.resolve_undefined(), @@ -597,56 +765,23 @@ unsafe fn run_connection_query( sql_ptr: *const u8, params_f: f64, rows_as_array: bool, + force_prepared: bool, ) -> *mut Promise { let sql = read_sql(sql_ptr); let params = JsValue::from_bits(params_f.to_bits()); let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); + let request = QueryRequest::new(sql, param_values, rows_as_array, force_prepared); + let target = connection_target(conn_handle); let promise = JsPromise::new(); let raw = promise.as_raw(); spawn_blocking(move || { + let rows_as_array = request.rows_as_array; let outcome: Result = tokio::runtime::Handle::current().block_on(async move { - let wrapper = get_handle_mut::(conn_handle) - .ok_or_else(|| "Invalid connection handle".to_string())?; - let conn = wrapper - .connection - .as_mut() - .ok_or_else(|| "Connection already closed".to_string())?; - let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for p in ¶m_values { - q = match p { - ParamValue::Null => q.bind(Option::::None), - ParamValue::String(s) => q.bind(s.clone()), - ParamValue::Number(n) => q.bind(*n), - ParamValue::Int(i) => q.bind(*i), - ParamValue::Bool(b) => q.bind(*b), - }; - } - if is_select { - let rows = tokio::time::timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - q.fetch_all(conn), - ) - .await - .map_err(|_| "Query timed out".to_string())? - .map_err(|e| format!("Query failed: {}", e))?; - Ok(QueryOutcome::Rows(raws_from_mysql_rows(rows))) - } else { - let res = tokio::time::timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - q.execute(conn), - ) - .await - .map_err(|_| "Query timed out".to_string())? - .map_err(|e| format!("Query failed: {}", e))?; - Ok(QueryOutcome::Executed { - affected_rows: res.rows_affected(), - last_insert_id: res.last_insert_id(), - }) - } + let target = target.ok_or_else(|| "Invalid connection handle".to_string())?; + execute_query_on_target(target, &request).await }); match outcome { // #1824: build the JS result on the MAIN thread. outcome_to_jsvalue @@ -670,7 +805,7 @@ pub unsafe extern "C" fn js_mysql2_connection_query( sql_ptr: *const u8, params_f: f64, ) -> *mut Promise { - run_connection_query(conn_handle, sql_ptr, params_f, false) + run_connection_query(conn_handle, sql_ptr, params_f, false, false) } /// `connection.execute(sql, params) -> Promise<[rows, fields]>`. @@ -684,25 +819,40 @@ pub unsafe extern "C" fn js_mysql2_connection_execute( sql_ptr: *const u8, params_f: f64, ) -> *mut Promise { - run_connection_query(conn_handle, sql_ptr, params_f, false) + run_connection_query(conn_handle, sql_ptr, params_f, false, true) } fn run_simple_command(conn_handle: Handle, sql: &'static str) -> *mut Promise { + let target = connection_target(conn_handle); let promise = JsPromise::new(); let raw = promise.as_raw(); spawn_blocking(move || { let result = tokio::runtime::Handle::current().block_on(async move { - let wrapper = get_handle_mut::(conn_handle) - .ok_or_else(|| "Invalid connection handle".to_string())?; - let conn = wrapper - .connection - .as_mut() - .ok_or_else(|| "Connection already closed".to_string())?; - sqlx::query(sql) - .execute(conn) - .await - .map(|_| ()) - .map_err(|e| format!("{}: {}", sql, e)) + let target = target.ok_or_else(|| "Invalid connection handle".to_string())?; + match target { + MysqlConnectionTarget::Direct(connection) => { + let mut slot = connection.lock().await; + let conn = slot + .as_mut() + .ok_or_else(|| "Connection already closed".to_string())?; + sqlx::raw_sql(sql) + .execute(conn) + .await + .map(|_| ()) + .map_err(|e| format!("{}: {}", sql, e)) + } + MysqlConnectionTarget::Pool(connection) => { + let mut slot = connection.lock().await; + let conn = slot + .as_mut() + .ok_or_else(|| "Pool connection released".to_string())?; + sqlx::raw_sql(sql) + .execute(&mut **conn) + .await + .map(|_| ()) + .map_err(|e| format!("{}: {}", sql, e)) + } + } }); match result { Ok(()) => promise.resolve_undefined(), @@ -712,6 +862,15 @@ fn run_simple_command(conn_handle: Handle, sql: &'static str) -> *mut Promise { raw } +fn transaction_sql_for_method(method: &str) -> Option<&'static str> { + match method { + "beginTransaction" => Some("START TRANSACTION"), + "commit" => Some("COMMIT"), + "rollback" => Some("ROLLBACK"), + _ => None, + } +} + #[no_mangle] pub extern "C" fn js_mysql2_connection_begin_transaction(conn_handle: Handle) -> *mut Promise { run_simple_command(conn_handle, "START TRANSACTION") @@ -740,13 +899,13 @@ impl MysqlPoolHandle { } pub struct MysqlPoolConnectionHandle { - pub connection: Option>, + pub connection: Arc>>>, } impl MysqlPoolConnectionHandle { pub fn new(conn: PoolConnection) -> Self { Self { - connection: Some(conn), + connection: Arc::new(Mutex::new(Some(conn))), } } } @@ -898,9 +1057,9 @@ unsafe extern "C" fn js_mysql2_handle_method_dispatch( }; // Only claim methods for handles we actually own. - let is_pool = perry_ffi::get_handle::(handle).is_some(); - let is_pool_conn = perry_ffi::get_handle::(handle).is_some(); - let is_conn = perry_ffi::get_handle::(handle).is_some(); + let is_pool = with_handle::(handle, |_| ()).is_some(); + let is_pool_conn = with_handle::(handle, |_| ()).is_some(); + let is_conn = with_handle::(handle, |_| ()).is_some(); if !is_pool && !is_pool_conn && !is_conn { return 0; } @@ -914,12 +1073,13 @@ unsafe extern "C" fn js_mysql2_handle_method_dispatch( .get(1) .copied() .unwrap_or(f64::from_bits(DISPATCH_TAG_UNDEFINED)); + let force_prepared = method == "execute"; let promise = if is_pool { - run_pool_query(handle, sql_ptr, params_f, rows_as_array) + run_pool_query(handle, sql_ptr, params_f, rows_as_array, force_prepared) } else if is_pool_conn { - run_pool_conn_query(handle, sql_ptr, params_f, rows_as_array) + run_pool_conn_query(handle, sql_ptr, params_f, rows_as_array, force_prepared) } else { - run_connection_query(handle, sql_ptr, params_f, rows_as_array) + run_connection_query(handle, sql_ptr, params_f, rows_as_array, force_prepared) }; dispatch_nanbox_ptr(promise) } @@ -929,6 +1089,12 @@ unsafe extern "C" fn js_mysql2_handle_method_dispatch( js_mysql2_pool_connection_release(handle); f64::from_bits(DISPATCH_TAG_UNDEFINED) } + method if (is_pool_conn || is_conn) && transaction_sql_for_method(method).is_some() => { + let Some(sql) = transaction_sql_for_method(method) else { + return 0; + }; + dispatch_nanbox_ptr(run_simple_command(handle, sql)) + } "end" if is_conn => dispatch_nanbox_ptr(js_mysql2_connection_end(handle)), // `mysql2/promise` pools are already promise-based: `pool.promise()` // returns the pool itself. Drizzle's `isCallbackClient` only reaches this @@ -963,52 +1129,32 @@ unsafe fn run_pool_query( sql_ptr: *const u8, params_f: f64, rows_as_array: bool, + force_prepared: bool, ) -> *mut Promise { let sql = read_sql(sql_ptr); let params = JsValue::from_bits(params_f.to_bits()); let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); + let request = QueryRequest::new(sql, param_values, rows_as_array, force_prepared); + let pool = with_handle::(pool_handle, |wrapper| wrapper.pool.clone()); let promise = JsPromise::new(); let raw = promise.as_raw(); spawn_blocking(move || { + let rows_as_array = request.rows_as_array; let outcome: Result = tokio::runtime::Handle::current().block_on(async move { - let wrapper = get_handle_mut::(pool_handle) - .ok_or_else(|| "Invalid pool handle".to_string())?; - let pool = &wrapper.pool; - let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for p in ¶m_values { - q = match p { - ParamValue::Null => q.bind(Option::::None), - ParamValue::String(s) => q.bind(s.clone()), - ParamValue::Number(n) => q.bind(*n), - ParamValue::Int(i) => q.bind(*i), - ParamValue::Bool(b) => q.bind(*b), - }; - } - if is_select { - let rows = tokio::time::timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - q.fetch_all(pool), - ) - .await - .map_err(|_| "Query timed out".to_string())? - .map_err(|e| format!("Query failed: {}", e))?; - Ok(QueryOutcome::Rows(raws_from_mysql_rows(rows))) - } else { - let res = tokio::time::timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - q.execute(pool), - ) - .await - .map_err(|_| "Query timed out".to_string())? - .map_err(|e| format!("Query failed: {}", e))?; - Ok(QueryOutcome::Executed { - affected_rows: res.rows_affected(), - last_insert_id: res.last_insert_id(), - }) - } + let pool = pool.ok_or_else(|| "Invalid pool handle".to_string())?; + // Explicitly check out one connection for the whole request so + // statement preparation, bind encoding, execution, and result + // draining cannot be split across independent pool operations. + let mut conn = tokio::time::timeout( + Duration::from_secs(DEFAULT_ACQUIRE_TIMEOUT_SECS), + pool.acquire(), + ) + .await + .map_err(|_| "Pool acquire timed out".to_string())? + .map_err(|e| format!("Pool acquire failed: {}", e))?; + execute_query_on_connection(&mut conn, &request).await }); match outcome { // #1824: build the JS result on the MAIN thread. outcome_to_jsvalue @@ -1030,7 +1176,7 @@ pub unsafe extern "C" fn js_mysql2_pool_query( sql_ptr: *const u8, params_f: f64, ) -> *mut Promise { - run_pool_query(pool_handle, sql_ptr, params_f, false) + run_pool_query(pool_handle, sql_ptr, params_f, false, false) } /// # Safety @@ -1041,20 +1187,20 @@ pub unsafe extern "C" fn js_mysql2_pool_execute( sql_ptr: *const u8, params_f: f64, ) -> *mut Promise { - run_pool_query(pool_handle, sql_ptr, params_f, false) + run_pool_query(pool_handle, sql_ptr, params_f, false, true) } #[no_mangle] pub extern "C" fn js_mysql2_pool_get_connection(pool_handle: Handle) -> *mut Promise { + let pool = with_handle::(pool_handle, |wrapper| wrapper.pool.clone()); let promise = JsPromise::new(); let raw = promise.as_raw(); spawn_blocking(move || { let result = tokio::runtime::Handle::current().block_on(async move { - let wrapper = get_handle_mut::(pool_handle) - .ok_or_else(|| "Invalid pool handle".to_string())?; + let pool = pool.ok_or_else(|| "Invalid pool handle".to_string())?; tokio::time::timeout( Duration::from_secs(DEFAULT_ACQUIRE_TIMEOUT_SECS), - wrapper.pool.acquire(), + pool.acquire(), ) .await .map_err(|_| "Pool acquire timed out".to_string())? @@ -1075,7 +1221,15 @@ pub extern "C" fn js_mysql2_pool_get_connection(pool_handle: Handle) -> *mut Pro /// underlying `PoolConnection` returns to the pool via Drop. #[no_mangle] pub extern "C" fn js_mysql2_pool_connection_release(conn_handle: Handle) { - take_handle::(conn_handle); + if let Some(wrapper) = take_handle::(conn_handle) { + // A query already in flight owns another Arc and holds this mutex. Wait + // for it to finish before dropping the checkout back into the pool. + spawn_blocking(move || { + tokio::runtime::Handle::current().block_on(async move { + wrapper.connection.lock().await.take(); + }); + }); + } } unsafe fn run_pool_conn_query( @@ -1083,55 +1237,29 @@ unsafe fn run_pool_conn_query( sql_ptr: *const u8, params_f: f64, rows_as_array: bool, + force_prepared: bool, ) -> *mut Promise { let sql = read_sql(sql_ptr); let params = JsValue::from_bits(params_f.to_bits()); let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); + let request = QueryRequest::new(sql, param_values, rows_as_array, force_prepared); + let connection = with_handle::(conn_handle, |wrapper| { + Arc::clone(&wrapper.connection) + }); let promise = JsPromise::new(); let raw = promise.as_raw(); spawn_blocking(move || { + let rows_as_array = request.rows_as_array; let outcome: Result = tokio::runtime::Handle::current().block_on(async move { - let wrapper = get_handle_mut::(conn_handle) - .ok_or_else(|| "Invalid pool-connection handle".to_string())?; - let conn = wrapper - .connection + let connection = + connection.ok_or_else(|| "Invalid pool-connection handle".to_string())?; + let mut slot = connection.lock().await; + let conn = slot .as_mut() .ok_or_else(|| "Pool connection released".to_string())?; - let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for p in ¶m_values { - q = match p { - ParamValue::Null => q.bind(Option::::None), - ParamValue::String(s) => q.bind(s.clone()), - ParamValue::Number(n) => q.bind(*n), - ParamValue::Int(i) => q.bind(*i), - ParamValue::Bool(b) => q.bind(*b), - }; - } - if is_select { - let rows = tokio::time::timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - q.fetch_all(&mut **conn), - ) - .await - .map_err(|_| "Query timed out".to_string())? - .map_err(|e| format!("Query failed: {}", e))?; - Ok(QueryOutcome::Rows(raws_from_mysql_rows(rows))) - } else { - let res = tokio::time::timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - q.execute(&mut **conn), - ) - .await - .map_err(|_| "Query timed out".to_string())? - .map_err(|e| format!("Query failed: {}", e))?; - Ok(QueryOutcome::Executed { - affected_rows: res.rows_affected(), - last_insert_id: res.last_insert_id(), - }) - } + execute_query_on_connection(conn, &request).await }); match outcome { // #1824: build the JS result on the MAIN thread. outcome_to_jsvalue @@ -1153,7 +1281,7 @@ pub unsafe extern "C" fn js_mysql2_pool_connection_query( sql_ptr: *const u8, params_f: f64, ) -> *mut Promise { - run_pool_conn_query(conn_handle, sql_ptr, params_f, false) + run_pool_conn_query(conn_handle, sql_ptr, params_f, false, false) } /// # Safety @@ -1164,7 +1292,7 @@ pub unsafe extern "C" fn js_mysql2_pool_connection_execute( sql_ptr: *const u8, params_f: f64, ) -> *mut Promise { - run_pool_conn_query(conn_handle, sql_ptr, params_f, false) + run_pool_conn_query(conn_handle, sql_ptr, params_f, false, true) } #[cfg(test)] @@ -1235,4 +1363,87 @@ mod tests { assert!(!is_row_returning_query("UPDATE t SET x = 1")); assert!(!is_row_returning_query("DELETE FROM t")); } + + #[test] + fn query_request_keeps_each_statement_with_its_own_params() { + let ddl = QueryRequest::new("DROP TABLE IF EXISTS t".into(), Vec::new(), false, false); + let insert = QueryRequest::new( + "INSERT INTO t (name, cents) VALUES (?, ?)".into(), + vec![ParamValue::String("x".into()), ParamValue::Int(100)], + false, + true, + ); + let select = QueryRequest::new( + "SELECT * FROM t WHERE id = ?".into(), + vec![ParamValue::Int(1)], + false, + true, + ); + + assert_eq!(ddl.params, Vec::::new()); + assert_eq!(insert.params.len(), 2); + assert_eq!(select.params, vec![ParamValue::Int(1)]); + assert!(!ddl.uses_prepared_statement()); + assert!(insert.uses_prepared_statement()); + assert!(select.uses_prepared_statement()); + } + + #[test] + fn execute_stays_prepared_even_without_params() { + let execute = QueryRequest::new("SELECT 1".into(), Vec::new(), false, true); + assert!(execute.uses_prepared_statement()); + } + + #[test] + fn both_connection_handle_families_resolve_to_serialized_targets() { + let direct_connection = Arc::new(Mutex::new(None)); + let direct_handle = register_handle(MysqlConnectionHandle { + connection: Arc::clone(&direct_connection), + }); + let pool_connection = Arc::new(Mutex::new(None)); + let pool_handle = register_handle(MysqlPoolConnectionHandle { + connection: Arc::clone(&pool_connection), + }); + + match connection_target(direct_handle) { + Some(MysqlConnectionTarget::Direct(resolved)) => { + assert!(Arc::ptr_eq(&resolved, &direct_connection)); + let _guard = resolved + .try_lock() + .expect("first operation locks connection"); + assert!( + direct_connection.try_lock().is_err(), + "a second operation on the same connection must serialize" + ); + } + _ => panic!("direct connection handle was not resolved"), + } + match connection_target(pool_handle) { + Some(MysqlConnectionTarget::Pool(resolved)) => { + assert!(Arc::ptr_eq(&resolved, &pool_connection)); + let _guard = resolved + .try_lock() + .expect("first operation locks pool connection"); + assert!( + pool_connection.try_lock().is_err(), + "a second operation on the same checkout must serialize" + ); + } + _ => panic!("pool connection handle was not resolved"), + } + + take_handle::(direct_handle); + take_handle::(pool_handle); + } + + #[test] + fn pool_connections_expose_the_full_transaction_command_set() { + assert_eq!( + transaction_sql_for_method("beginTransaction"), + Some("START TRANSACTION") + ); + assert_eq!(transaction_sql_for_method("commit"), Some("COMMIT")); + assert_eq!(transaction_sql_for_method("rollback"), Some("ROLLBACK")); + assert_eq!(transaction_sql_for_method("release"), None); + } } diff --git a/crates/perry-ext-mysql2/src/test_async_shims.rs b/crates/perry-ext-mysql2/src/test_async_shims.rs new file mode 100644 index 0000000000..47945057b6 --- /dev/null +++ b/crates/perry-ext-mysql2/src/test_async_shims.rs @@ -0,0 +1,103 @@ +//! Test-only host shims for the standalone extension test binary. +//! +//! Production binaries receive these symbols from perry-stdlib's async bridge. + +use perry_ffi::{NativeAsyncCompletion, Promise}; +use std::ffi::c_void; + +#[no_mangle] +pub extern "C" fn perry_ffi_promise_new() -> *mut Promise { + perry_runtime::promise::js_promise_new() as *mut Promise +} + +#[no_mangle] +pub extern "C" fn perry_ffi_promise_resolve_bits(promise: *mut Promise, bits: u64) { + perry_runtime::promise::js_promise_resolve( + promise as *mut perry_runtime::Promise, + f64::from_bits(bits), + ); +} + +#[no_mangle] +pub extern "C" fn perry_ffi_promise_reject_bits(promise: *mut Promise, bits: u64) { + perry_runtime::promise::js_promise_reject( + promise as *mut perry_runtime::Promise, + f64::from_bits(bits), + ); +} + +#[no_mangle] +pub extern "C" fn perry_ffi_promise_resolve_deferred( + promise: *mut Promise, + ctx: *mut c_void, + invoke: extern "C" fn(*mut c_void) -> u64, +) { + perry_ffi_promise_resolve_bits(promise, invoke(ctx)); +} + +#[no_mangle] +pub extern "C" fn perry_ffi_spawn_blocking(ctx: *mut c_void, invoke: extern "C" fn(*mut c_void)) { + invoke(ctx); +} + +#[no_mangle] +pub extern "C" fn perry_ffi_spawn_blocking_with_reactor( + ctx: *mut c_void, + invoke: extern "C" fn(*mut c_void), +) { + invoke(ctx); +} + +#[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) {} diff --git a/crates/perry-runtime/src/object/class_registry/state.rs b/crates/perry-runtime/src/object/class_registry/state.rs index 9baf2e4acd..1c655c8b0a 100644 --- a/crates/perry-runtime/src/object/class_registry/state.rs +++ b/crates/perry-runtime/src/object/class_registry/state.rs @@ -750,30 +750,16 @@ pub(crate) fn class_decl_prototype_value(class_id: u32) -> f64 { if parent.is_pointer() { let parent_addr = parent.as_pointer::() as usize; if crate::closure::is_closure_ptr(parent_addr) { + // Use the same observable `.prototype` read as ordinary + // property access. Plain functions and bound native-module + // constructor exports materialize this object lazily, while + // explicit, deleted, and generator prototypes must retain + // their own semantics. let parent_proto = - crate::closure::closure_get_dynamic_prop(parent_addr, "prototype"); - // A bound native-module constructor export imported directly - // (`import { EventEmitter } from "events"; class X extends - // EventEmitter {}`) carries `.prototype` only lazily: the raw - // dynamic-slot read above is still `undefined` because the - // synthetic prototype object (which carries the EventEmitter - // method surface) is materialized on demand, not at closure - // mint time. Resolve it exactly as an ordinary `Y.prototype` - // property read does, so the `extends` edge links to that real - // prototype instead of throwing. `Stream` and the net/http - // server classes share this shape. A non-constructor bound - // method still resolves to `None` here, so a genuinely - // prototype-less parent (e.g. a bare `fn.bind(...)`) still - // throws below, matching Node. - let resolved_proto = if class_parent_prototype_bits(parent_proto).is_some() { - parent_proto - } else { - super::function_prototype::ordinary_function_prototype_value_for_read( + super::function_prototype::js_function_prototype_value_for_read( dynamic_parent.get_nanbox_f64(), - ) - .unwrap_or(parent_proto) - }; - if let Some(bits) = class_parent_prototype_bits(resolved_proto) { + ); + if let Some(bits) = class_parent_prototype_bits(parent_proto) { Some(bits) } else { super::super::object_ops::throw_object_type_error( 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}"); +} diff --git a/crates/perry/tests/versioned_indexed_loop_forwarding.rs b/crates/perry/tests/versioned_indexed_loop_forwarding.rs new file mode 100644 index 0000000000..e7559d7caf --- /dev/null +++ b/crates/perry/tests/versioned_indexed_loop_forwarding.rs @@ -0,0 +1,108 @@ +//! Runtime regression for forwarded arrays in fallback-free checked-reader +//! loops. Array growth preserves JavaScript identity by leaving a forwarding +//! stub behind; loop admission must normalize one edge to the live array, and +//! a later callback-driven growth must still side-exit before the next effect. + +use std::path::PathBuf; +use std::process::{Command, Output}; + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +fn run_fixture(binary: &std::path::Path, force_evacuation: bool) -> Output { + let mut command = Command::new(binary); + if force_evacuation { + command.env("PERRY_GC_FORCE_EVACUATE", "1"); + } else { + command.env_remove("PERRY_GC_FORCE_EVACUATE"); + } + command + .output() + .expect("run versioned indexed-loop fixture") +} + +#[test] +fn forwarded_arrays_enter_safely_and_callback_growth_resumes_generically() { + let dir = tempfile::tempdir().expect("tempdir"); + let entry = dir.path().join("main.ts"); + let binary = dir.path().join("main_bin"); + std::fs::write( + &entry, + r#" +class Reader { + entities: number[] = []; + + private checkedRead(column: any[] | undefined, index: number, type: number): any { + if (column === undefined) throw new Error("missing column " + type); + const value = column[index]; + if (value === 99) throw new Error("missing value " + type + " at " + index); + return value; + } + + iterate( + column: any[], + callback: (entity: number, value: any) => void, + entityFilter?: (entity: number) => boolean, + ): void { + const entities = this.entities; + const entityCount = entities.length; + const cb = callback; + for (let i = 0; i < entityCount; i++) { + const entity = entities[i]!; + if (entityFilter && !entityFilter(entity)) continue; + cb(entity, this.checkedRead(column, i, 1)); + } + } +} + +const reader = new Reader(); +const column: any[] = []; +for (let i = 0; i < 4096; i++) { + reader.entities.push(i); + column.push({ n: i }); +} + +let sum = 0; +let grew = false; +reader.iterate(column, (entity, value) => { + sum += entity + value.n; + if (!grew) { + grew = true; + for (let i = 0; i < 4096; i++) column.push({ n: -1 }); + } +}, undefined); + +console.log(sum + ":" + reader.entities.length + ":" + column.length); +"#, + ) + .expect("write versioned indexed-loop fixture"); + + let compile = Command::new(perry_bin()) + .current_dir(dir.path()) + .arg("compile") + .arg(&entry) + .arg("-o") + .arg(&binary) + .arg("--no-cache") + .arg("--no-auto-optimize") + .output() + .expect("compile versioned indexed-loop fixture"); + assert!( + compile.status.success(), + "compile failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&compile.stdout), + String::from_utf8_lossy(&compile.stderr) + ); + + for force_evacuation in [false, true] { + let run = run_fixture(&binary, force_evacuation); + assert!( + run.status.success(), + "fixture failed (force_evacuation={force_evacuation})\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&run.stdout), + String::from_utf8_lossy(&run.stderr) + ); + assert_eq!(String::from_utf8_lossy(&run.stdout), "16773120:4096:8192\n"); + } +} diff --git a/test-files/test_issue_8745_8746_mysql2_operation_isolation.ts b/test-files/test_issue_8745_8746_mysql2_operation_isolation.ts new file mode 100644 index 0000000000..eb53011232 --- /dev/null +++ b/test-files/test_issue_8745_8746_mysql2_operation_isolation.ts @@ -0,0 +1,69 @@ +// parity-skip: requires a live MySQL fixture; native wrapper unit-tested +// Regression coverage for issues #8745 and #8746. +// +// Run against a local MySQL database after removing the skip marker. The loop +// mixes text-protocol DDL with prepared statements of different arities; every +// statement must retain its own parameter vector. The checked-out connection +// then exercises the canonical row-lock transaction lifecycle. +// +// platforms: skip + +import mysql from 'mysql2/promise'; + +const pool = mysql.createPool({ + host: 'localhost', port: 3306, user: 'perry', password: 'perry', + database: 'perry_hub', +}); + +async function main(): Promise { + for (let round = 0; round < 25; round++) { + await pool.query('DROP TABLE IF EXISTS perry_issue_8745'); + await pool.query( + 'CREATE TABLE perry_issue_8745 (' + + 'id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50), cents INT)', + ); + await pool.execute( + 'INSERT INTO perry_issue_8745 (name, cents) VALUES (?, ?)', + ['round-' + round, 100 + round], + ); + const selected: any = await pool.execute( + 'SELECT * FROM perry_issue_8745 WHERE id = ?', + [1], + ); + if (selected[0][0].cents !== 100 + round) { + throw new Error('wrong prepared-statement parameters in round ' + round); + } + } + + const connection = await pool.getConnection(); + try { + await connection.beginTransaction(); + const locked: any = await connection.execute( + 'SELECT cents FROM perry_issue_8745 WHERE id = ? FOR UPDATE', + [1], + ); + await connection.execute( + 'UPDATE perry_issue_8745 SET cents = ? WHERE id = ?', + [locked[0][0].cents + 1, 1], + ); + await connection.commit(); + + await connection.beginTransaction(); + await connection.execute( + 'UPDATE perry_issue_8745 SET cents = ? WHERE id = ?', + [9999, 1], + ); + await connection.rollback(); + } finally { + connection.release(); + } + + const finalRows: any = await pool.execute( + 'SELECT cents FROM perry_issue_8745 WHERE id = ?', + [1], + ); + console.log(finalRows[0][0].cents); + await pool.end(); +} + +main();