diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index dde59c69f6..949f1cb142 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -391,6 +391,16 @@ jobs: python3 scripts/gc_pin_sites.py --self-test python3 scripts/gc_pin_sites.py + # #9552. A promise handed to native code as a bare address is invisible + # to every root scanner until its completion is queued back. The + # cross-thread constructor pins it for that window; the arena constructor + # cannot. This finds arena promises reaching a native settlement sink. + - name: Cross-thread promise provenance (#9552) + if: ${{ !cancelled() }} + run: | + python3 scripts/check_cross_thread_promise_provenance.py --self-test + python3 scripts/check_cross_thread_promise_provenance.py + # #7231. A runtime-side table holding a GC pointer IS a root, and nothing # static could see that class before: gc_root_dominance_check.py reads # emitted LLVM IR and a thread_local is not in it. #7226, #7239, #7268 and diff --git a/crates/perry-codegen/src/lower_call/native_table/databases.rs b/crates/perry-codegen/src/lower_call/native_table/databases.rs index 31e19161e8..5c169be374 100644 --- a/crates/perry-codegen/src/lower_call/native_table/databases.rs +++ b/crates/perry-codegen/src/lower_call/native_table/databases.rs @@ -45,7 +45,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "query", class_filter: Some("Pool"), runtime: "js_mysql2_pool_query", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -54,7 +54,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "execute", class_filter: Some("Pool"), runtime: "js_mysql2_pool_execute", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -72,7 +72,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "query", class_filter: Some("Pool"), runtime: "js_mysql2_pool_query", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -81,7 +81,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "execute", class_filter: Some("Pool"), runtime: "js_mysql2_pool_execute", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -100,7 +100,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "query", class_filter: Some("PoolConnection"), runtime: "js_mysql2_pool_connection_query", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -109,7 +109,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "execute", class_filter: Some("PoolConnection"), runtime: "js_mysql2_pool_connection_execute", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -118,7 +118,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "query", class_filter: Some("PoolConnection"), runtime: "js_mysql2_pool_connection_query", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -127,7 +127,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "execute", class_filter: Some("PoolConnection"), runtime: "js_mysql2_pool_connection_execute", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, // mysql2 generic instance methods (Connection fallback, class_filter: None) @@ -137,7 +137,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "query", class_filter: None, runtime: "js_mysql2_connection_query", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -146,7 +146,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "execute", class_filter: None, runtime: "js_mysql2_connection_execute", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -209,7 +209,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "query", class_filter: None, runtime: "js_mysql2_connection_query", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { @@ -218,7 +218,7 @@ pub(super) const DATABASES_ROWS: &[NativeModSig] = &[ method: "execute", class_filter: None, runtime: "js_mysql2_connection_execute", - args: &[NA_STR, NA_F64], + args: &[NA_F64, NA_F64], ret: NR_PTR, }, NativeModSig { diff --git a/crates/perry-ext-mysql2/src/lib.rs b/crates/perry-ext-mysql2/src/lib.rs index 3a21c0a6b0..c700cc99e9 100644 --- a/crates/perry-ext-mysql2/src/lib.rs +++ b/crates/perry-ext-mysql2/src/lib.rs @@ -18,7 +18,7 @@ use perry_ffi::{ alloc_string, build_object_shape, js_array_alloc, js_array_get, js_array_length, js_array_push, js_object_alloc_with_shape, js_object_get_field, js_object_set_field, register_handle, spawn_blocking, take_handle, value_byte_slice, with_handle, ArrayHeader, Handle, JsPromise, - JsValue, ObjectHeader, Promise, StringHeader, SHORT_STRING_MAX_LEN, + JsValue, ObjectHeader, Promise, StringHeader, TransientRootScope, SHORT_STRING_MAX_LEN, }; use sqlx::mysql::{MySqlConnection, MySqlDatabaseError, MySqlPool, MySqlPoolOptions, MySqlRow}; use sqlx::pool::PoolConnection; @@ -694,17 +694,6 @@ fn rejected_params_promise(message: String) -> *mut Promise { raw } -unsafe fn read_sql(sql_ptr: *const u8) -> String { - if sql_ptr.is_null() { - return String::new(); - } - let header = sql_ptr as *const StringHeader; - let len = (*header).byte_len as usize; - let data = sql_ptr.add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data, len); - std::str::from_utf8(bytes).unwrap_or("").to_string() -} - // ── Connection ──────────────────────────────────────────────────── pub struct MysqlConnectionHandle { @@ -988,18 +977,14 @@ pub extern "C" fn js_mysql2_connection_end(conn_handle: Handle) -> *mut Promise unsafe fn run_connection_query( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, 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 = match extract_params_from_jsvalue(params) { - Ok(values) => values, + let request = match parse_query_request(query_f, params_f, force_prepared) { + Ok(request) => request, Err(message) => return rejected_params_promise(message), }; - let request = QueryRequest::new(sql, param_values, rows_as_array, force_prepared); let target = connection_target(conn_handle); let promise = JsPromise::new(); @@ -1028,28 +1013,28 @@ unsafe fn run_connection_query( /// `connection.query(sql, params) -> Promise<[rows, fields]>`. /// /// # Safety -/// `sql_ptr` must be null or a Perry-runtime `StringHeader`. +/// `query_f` must be a SQL string or mysql2 query-options object. #[no_mangle] pub unsafe extern "C" fn js_mysql2_connection_query( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, ) -> *mut Promise { - run_connection_query(conn_handle, sql_ptr, params_f, false, false) + run_connection_query(conn_handle, query_f, params_f, false) } /// `connection.execute(sql, params) -> Promise<[rows, fields]>`. /// Same backing as `query` for now (sqlx prepares all queries). /// /// # Safety -/// `sql_ptr` must be null or a Perry-runtime `StringHeader`. +/// `query_f` must be a SQL string or mysql2 query-options object. #[no_mangle] pub unsafe extern "C" fn js_mysql2_connection_execute( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, ) -> *mut Promise { - run_connection_query(conn_handle, sql_ptr, params_f, false, true) + run_connection_query(conn_handle, query_f, params_f, true) } fn run_simple_command(conn_handle: Handle, sql: &'static str) -> *mut Promise { @@ -1206,6 +1191,14 @@ extern "C" { fn js_register_handle_method_dispatch_extension( f: unsafe extern "C" fn(i64, *const u8, usize, *const f64, usize, *mut f64) -> i32, ); + fn js_register_handle_property_dispatch_extension( + f: unsafe extern "C" fn(i64, *const u8, usize, *mut f64) -> i32, + ); + fn js_class_method_bind( + instance: f64, + method_name_ptr: *const u8, + method_name_len: usize, + ) -> f64; // Runtime generic field read; returns the runtime `JSValue` (repr-transparent // u64), ABI-compatible with `u64` here. fn js_object_get_field_by_name(obj: *const ObjectHeader, key: *const StringHeader) -> u64; @@ -1219,43 +1212,69 @@ fn ensure_dispatch_registered() { static REGISTER: std::sync::Once = std::sync::Once::new(); REGISTER.call_once(|| unsafe { js_register_handle_method_dispatch_extension(js_mysql2_handle_method_dispatch); + js_register_handle_property_dispatch_extension(js_mysql2_handle_property_dispatch); }); } /// Read a named field off a JS object value. Returns `JsValue::UNDEFINED` for a /// non-object receiver or a missing key. unsafe fn object_field_by_name(obj: JsValue, name: &str) -> JsValue { - let obj_ptr = obj.as_pointer::(); + let roots = TransientRootScope::enter(); + let obj = roots.root_nanbox(f64::from_bits(obj.bits())); + let key = alloc_string(name); + let obj_ptr = JsValue::from_bits(obj.get().to_bits()).as_pointer::(); if obj_ptr.is_null() { return JsValue::UNDEFINED; } - let key = alloc_string(name); let bits = js_object_get_field_by_name(obj_ptr, key.as_raw()); JsValue::from_bits(bits) } -/// Resolve the SQL `StringHeader` pointer and `rowsAsArray` flag from a `query` -/// argument that is either a SQL string or a mysql2 options object -/// (`{ sql, rowsAsArray? }` — Drizzle's shape). -unsafe fn query_sql_ptr_and_rows_as_array(arg: JsValue) -> Option<(*const u8, bool)> { - if arg.is_string() { - let p = arg.as_string_ptr(); - if p.is_null() { - return None; - } - return Some((p as *const u8, false)); - } - if arg.is_pointer() { - let sql_val = object_field_by_name(arg, "sql"); - if sql_val.is_string() { - let p = sql_val.as_string_ptr(); - if !p.is_null() { - let rows_as_array = object_field_by_name(arg, "rowsAsArray").to_bool(); - return Some((p as *const u8, rows_as_array)); - } - } - } - None +/// Parse both mysql2 query signatures while the JS arguments are rooted. +unsafe fn parse_query_request( + query_f: f64, + params_f: f64, + force_prepared: bool, +) -> Result { + let roots = TransientRootScope::enter(); + let query = roots.root_nanbox(query_f); + let supplied_params = roots.root_nanbox(params_f); + let query_value = JsValue::from_bits(query.get().to_bits()); + let (sql, rows_as_array, option_values) = if query_value.is_any_string() { + ( + jsvalue_to_string(query_value).unwrap_or_default(), + false, + JsValue::UNDEFINED, + ) + } else if query_value.is_pointer() { + let sql = jsvalue_to_string(object_field_by_name( + JsValue::from_bits(query.get().to_bits()), + "sql", + )) + .ok_or_else(|| "Query options must include a SQL string".to_string())?; + let rows = object_field_by_name(JsValue::from_bits(query.get().to_bits()), "rowsAsArray"); + ( + sql, + rows.is_bool() && rows.to_bool(), + object_field_by_name(JsValue::from_bits(query.get().to_bits()), "values"), + ) + } else { + return Err("Query must be a SQL string or options object".to_string()); + }; + + let supplied_params = JsValue::from_bits(supplied_params.get().to_bits()); + let params = if supplied_params.is_undefined() { + option_values + } else { + supplied_params + }; + let params = extract_params_from_jsvalue(params)?; + Ok(QueryRequest::new( + sql, + params, + rows_as_array, + force_prepared, + )) } /// Handle-method dispatch extension for mysql2 pool / connection handles. @@ -1281,11 +1300,10 @@ unsafe extern "C" fn js_mysql2_handle_method_dispatch( } else { std::slice::from_raw_parts(args_ptr, args_len) }; - let arg = |i: usize| -> JsValue { + let arg = |i: usize| -> f64 { args.get(i) .copied() - .map(|f| JsValue::from_bits(f.to_bits())) - .unwrap_or(JsValue::UNDEFINED) + .unwrap_or(f64::from_bits(DISPATCH_TAG_UNDEFINED)) }; // Only claim methods for handles we actually own. @@ -1298,20 +1316,17 @@ unsafe extern "C" fn js_mysql2_handle_method_dispatch( let result: f64 = match method { "query" | "execute" => { - let Some((sql_ptr, rows_as_array)) = query_sql_ptr_and_rows_as_array(arg(0)) else { - return 0; - }; let params_f = args .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, force_prepared) + run_pool_query(handle, arg(0), params_f, force_prepared) } else if is_pool_conn { - run_pool_conn_query(handle, sql_ptr, params_f, rows_as_array, force_prepared) + run_pool_conn_query(handle, arg(0), params_f, force_prepared) } else { - run_connection_query(handle, sql_ptr, params_f, rows_as_array, force_prepared) + run_connection_query(handle, arg(0), params_f, force_prepared) }; dispatch_nanbox_ptr(promise) } @@ -1341,6 +1356,66 @@ unsafe extern "C" fn js_mysql2_handle_method_dispatch( 1 } +/// Reflect mysql2 methods as callable properties. Drizzle uses +/// `"getConnection" in client` to decide whether a transaction must check out +/// and pin a pool connection; method-call dispatch alone cannot satisfy that +/// probe. +#[no_mangle] +unsafe extern "C" fn js_mysql2_handle_property_dispatch( + handle: i64, + property_name_ptr: *const u8, + property_name_len: usize, + out: *mut f64, +) -> i32 { + if property_name_ptr.is_null() || property_name_len == 0 { + return 0; + } + let property = match std::str::from_utf8(std::slice::from_raw_parts( + property_name_ptr, + property_name_len, + )) { + Ok(property) => property, + Err(_) => return 0, + }; + 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(); + let available = (is_pool + && matches!( + property, + "query" | "execute" | "end" | "getConnection" | "promise" + )) + || (is_pool_conn + && matches!( + property, + "query" | "execute" | "release" | "beginTransaction" | "commit" | "rollback" + )) + || (is_conn + && matches!( + property, + "query" + | "execute" + | "end" + | "beginTransaction" + | "commit" + | "rollback" + | "promise" + )); + if !available { + return 0; + } + + let value = js_class_method_bind( + dispatch_nanbox_ptr(handle as *mut u8), + property.as_ptr(), + property.len(), + ); + if !out.is_null() { + *out = value; + } + 1 +} + #[no_mangle] pub extern "C" fn js_mysql2_pool_end(pool_handle: Handle) -> *mut Promise { let promise = JsPromise::new(); @@ -1358,18 +1433,14 @@ pub extern "C" fn js_mysql2_pool_end(pool_handle: Handle) -> *mut Promise { unsafe fn run_pool_query( pool_handle: Handle, - sql_ptr: *const u8, + query_f: f64, 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 = match extract_params_from_jsvalue(params) { - Ok(values) => values, + let request = match parse_query_request(query_f, params_f, force_prepared) { + Ok(request) => request, Err(message) => return rejected_params_promise(message), }; - 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(); @@ -1404,25 +1475,25 @@ unsafe fn run_pool_query( } /// # Safety -/// `sql_ptr` must be null or a Perry-runtime `StringHeader`. +/// `query_f` must be a SQL string or mysql2 query-options object. #[no_mangle] pub unsafe extern "C" fn js_mysql2_pool_query( pool_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, ) -> *mut Promise { - run_pool_query(pool_handle, sql_ptr, params_f, false, false) + run_pool_query(pool_handle, query_f, params_f, false) } /// # Safety -/// `sql_ptr` must be null or a Perry-runtime `StringHeader`. +/// `query_f` must be a SQL string or mysql2 query-options object. #[no_mangle] pub unsafe extern "C" fn js_mysql2_pool_execute( pool_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, ) -> *mut Promise { - run_pool_query(pool_handle, sql_ptr, params_f, false, true) + run_pool_query(pool_handle, query_f, params_f, true) } #[no_mangle] @@ -1469,18 +1540,14 @@ pub extern "C" fn js_mysql2_pool_connection_release(conn_handle: Handle) { unsafe fn run_pool_conn_query( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, 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 = match extract_params_from_jsvalue(params) { - Ok(values) => values, + let request = match parse_query_request(query_f, params_f, force_prepared) { + Ok(request) => request, Err(message) => return rejected_params_promise(message), }; - let request = QueryRequest::new(sql, param_values, rows_as_array, force_prepared); let connection = with_handle::(conn_handle, |wrapper| { Arc::clone(&wrapper.connection) }); @@ -1512,25 +1579,25 @@ unsafe fn run_pool_conn_query( } /// # Safety -/// `sql_ptr` must be null or a Perry-runtime `StringHeader`. +/// `query_f` must be a SQL string or mysql2 query-options object. #[no_mangle] pub unsafe extern "C" fn js_mysql2_pool_connection_query( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, ) -> *mut Promise { - run_pool_conn_query(conn_handle, sql_ptr, params_f, false, false) + run_pool_conn_query(conn_handle, query_f, params_f, false) } /// # Safety -/// `sql_ptr` must be null or a Perry-runtime `StringHeader`. +/// `query_f` must be a SQL string or mysql2 query-options object. #[no_mangle] pub unsafe extern "C" fn js_mysql2_pool_connection_execute( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, ) -> *mut Promise { - run_pool_conn_query(conn_handle, sql_ptr, params_f, false, true) + run_pool_conn_query(conn_handle, query_f, params_f, true) } #[cfg(test)] @@ -1810,7 +1877,7 @@ mod tests { let promise = unsafe { js_mysql2_connection_execute( perry_ffi::INVALID_HANDLE, - sql.as_raw() as *const u8, + f64::from_bits(JsValue::from_string_ptr(sql.as_raw()).bits()), f64::from_bits(JsValue::UNDEFINED.bits()), ) }; diff --git a/crates/perry-runtime/src/atomics.rs b/crates/perry-runtime/src/atomics.rs index 2b35827d6c..d2abd66ba1 100644 --- a/crates/perry-runtime/src/atomics.rs +++ b/crates/perry-runtime/src/atomics.rs @@ -768,11 +768,9 @@ pub extern "C" fn js_atomics_wait_async( // Cross-thread variant: referenced only by a raw usize in the pending // results queue until drained — must not live in the copying nursery // (the from-space flip ignores pins that no scanner reaches). + // #9552: the cross-thread constructor pins the promise until it settles; + // this only has to keep the event loop alive until the result lands. let promise = crate::promise::js_promise_new_cross_thread(); - // Pin the promise + keep the event loop alive until the async result lands. - unsafe { - crate::thread::pin_promise(promise); - } crate::thread::thread_job_begin(); let promise_usize = promise as usize; // #6185: the promise belongs to the agent calling `waitAsync`. The futex diff --git a/crates/perry-runtime/src/gc/mod.rs b/crates/perry-runtime/src/gc/mod.rs index ea794f7e55..655d8d4162 100644 --- a/crates/perry-runtime/src/gc/mod.rs +++ b/crates/perry-runtime/src/gc/mod.rs @@ -127,7 +127,7 @@ mod pin; pub(crate) use pin::test_reset_young_pin_latch; pub use pin::{ copied_minor_preflight_skips, copied_minor_preflight_walks, pin_object, pin_object_non_young, - unpin_object, + pin_user_ptr_non_young, unpin_object, unpin_user_ptr, }; use pin::{note_preflight_skipped, note_preflight_walked, young_pin_latch_armed}; /// Software prefetch helpers for the collector's pointer-chasing loops diff --git a/crates/perry-runtime/src/gc/pin.rs b/crates/perry-runtime/src/gc/pin.rs index 4b24e5232b..a64b53cc95 100644 --- a/crates/perry-runtime/src/gc/pin.rs +++ b/crates/perry-runtime/src/gc/pin.rs @@ -75,7 +75,7 @@ use std::cell::Cell; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use super::types::{GcHeader, GC_FLAG_ARENA, GC_FLAG_PINNED}; +use super::types::{GcHeader, GC_FLAG_ARENA, GC_FLAG_PINNED, GC_HEADER_SIZE}; crate::perry_thread_local! { static COPYING_WALK_PHASE: Cell> = @@ -266,6 +266,29 @@ pub(crate) unsafe fn pin_constrains_copying_minor_for_tests(header: *mut GcHeade /// # Safety /// /// As [`pin_object`]. +/// Pin the non-young object whose USER pointer is `user_ptr` (#9552). +/// +/// The address arithmetic lives here, next to the flag it serves, so callers +/// that hold a `*mut Promise` (or any other user pointer) do not each grow a +/// bare `GcHeader` cast. Malloc-resident and old-arena objects only: the +/// young-pin latch is deliberately not consulted (see `pin_object_non_young`). +#[inline] +pub unsafe fn pin_user_ptr_non_young(user_ptr: *mut u8) { + if user_ptr.is_null() { + return; + } + pin_object_non_young(user_ptr.sub(GC_HEADER_SIZE) as *mut GcHeader); +} + +/// Release the pin on the object whose USER pointer is `user_ptr` (#9552). +#[inline] +pub unsafe fn unpin_user_ptr(user_ptr: *mut u8) { + if user_ptr.is_null() { + return; + } + unpin_object(user_ptr.sub(GC_HEADER_SIZE) as *mut GcHeader); +} + #[inline] pub unsafe fn unpin_object(header: *mut GcHeader) { if header.is_null() { diff --git a/crates/perry-runtime/src/gc/tests/alloc.rs b/crates/perry-runtime/src/gc/tests/alloc.rs index ff33d88154..5c0ed4c297 100644 --- a/crates/perry-runtime/src/gc/tests/alloc.rs +++ b/crates/perry-runtime/src/gc/tests/alloc.rs @@ -639,6 +639,7 @@ fn alloc_malloc_kind_test_object(obj_type: u8) -> *mut u8 { std::ptr::write( ptr as *mut crate::promise::Promise, crate::promise::Promise { + native_pinned: 0, state: crate::promise::PromiseState::Pending, value: 0.0, reason: 0.0, diff --git a/crates/perry-runtime/src/gc/tests/support.rs b/crates/perry-runtime/src/gc/tests/support.rs index 6b35535846..130c0f7a08 100644 --- a/crates/perry-runtime/src/gc/tests/support.rs +++ b/crates/perry-runtime/src/gc/tests/support.rs @@ -63,6 +63,7 @@ pub(super) unsafe fn alloc_old_test_promise() -> *mut crate::promise::Promise { std::ptr::write( ptr, crate::promise::Promise { + native_pinned: 0, state: crate::promise::PromiseState::Pending, value: 0.0, reason: 0.0, @@ -784,6 +785,7 @@ pub(super) fn allocate_dead_malloc_churn_headers(per_type: usize) -> Vec std::ptr::write( ptr, crate::promise::Promise { + native_pinned: 0, state: crate::promise::PromiseState::Pending, value: 0.0, reason: 0.0, diff --git a/crates/perry-runtime/src/promise/cross_thread_pin_tests.rs b/crates/perry-runtime/src/promise/cross_thread_pin_tests.rs new file mode 100644 index 0000000000..5541d67bc4 --- /dev/null +++ b/crates/perry-runtime/src/promise/cross_thread_pin_tests.rs @@ -0,0 +1,128 @@ +//! #9552 — a promise whose address leaves the runtime as a bare `usize` (a +//! worker future, a pending-result queue, a native async token) is pinned by +//! `js_promise_new_cross_thread` and released by its settlement. These pin the +//! constructor/settlement contract and the trust boundary that classifies a +//! returning address. + +use super::native_async::{ + js_native_async_completion_new, js_native_async_completion_promise, + js_native_async_drop_promise_token, native_async_promise_has_token, test_native_async_lock, + test_reset_native_async_registry, +}; +use super::{ + classify_native_promise_addr, js_promise_new, js_promise_new_cross_thread, js_promise_reject, + js_promise_resolve, NativePromiseAddr, Promise, +}; +use crate::value::addr_class::try_read_gc_header; + +fn gc_flags(promise: *mut Promise) -> u8 { + unsafe { try_read_gc_header(promise as usize) } + .expect("a freshly minted promise is a tracked heap object") + .gc_flags +} + +fn pinned(promise: *mut Promise) -> bool { + gc_flags(promise) & crate::gc::GC_FLAG_PINNED != 0 +} + +#[test] +fn cross_thread_promise_is_pinned_at_creation_and_released_by_fulfilment() { + let _guard = test_native_async_lock(); + let promise = js_promise_new_cross_thread(); + assert_eq!( + gc_flags(promise) & crate::gc::GC_FLAG_ARENA, + 0, + "cross-thread promises are malloc-resident" + ); + assert!(pinned(promise), "#9552: the constructor takes the pin"); + assert_eq!(unsafe { (*promise).native_pinned }, 1); + + js_promise_resolve(promise, 1.0); + assert!(!pinned(promise), "settlement releases the pin"); + assert_eq!(unsafe { (*promise).native_pinned }, 0); + + // A second settlement is a no-op on an already-settled promise and must + // not touch the pin state. + js_promise_reject(promise, 2.0); + assert!(!pinned(promise)); +} + +#[test] +fn rejection_releases_the_pin_too() { + let _guard = test_native_async_lock(); + let promise = js_promise_new_cross_thread(); + assert!(pinned(promise)); + js_promise_reject(promise, 2.0); + assert!(!pinned(promise)); + assert_eq!(unsafe { (*promise).native_pinned }, 0); +} + +#[test] +fn cross_thread_promise_survives_a_full_collection_while_only_native_code_holds_it() { + let _guard = test_native_async_lock(); + // Hold the address the way a worker future does: as a bare integer no + // root scanner visits. XOR-hide it so a conservative stack scan (if one + // were to run) cannot keep the object alive by accident and make the + // assertion vacuous. + const MASK: usize = 0x5555_5555_5555_5555; + let hidden = (js_promise_new_cross_thread() as usize) ^ MASK; + crate::gc::js_gc_collect(); + let raw = hidden ^ MASK; + match classify_native_promise_addr(raw) { + NativePromiseAddr::Live(promise) => { + assert!(pinned(promise), "still pinned while in flight"); + js_promise_resolve(promise, 3.0); + assert!(!pinned(promise)); + } + other => panic!("#9552: in-flight promise did not survive the collection: {other:?}"), + } +} + +#[test] +fn arena_promises_carry_no_pin() { + let _guard = test_native_async_lock(); + let promise = js_promise_new(); + assert!(!pinned(promise)); + assert_eq!(unsafe { (*promise).native_pinned }, 0); + js_promise_resolve(promise, 1.0); + assert!(!pinned(promise)); +} + +#[test] +fn dropping_a_token_without_settling_releases_the_pin() { + let _guard = test_native_async_lock(); + test_reset_native_async_registry(); + let token = js_native_async_completion_new(0); + let promise = js_native_async_completion_promise(token); + assert!(pinned(promise), "token promises are cross-thread promises"); + assert!(native_async_promise_has_token(promise)); + // The token was the promise's root; once it is gone the pin must not keep + // a never-settling promise alive forever. + js_native_async_drop_promise_token(promise); + assert!(!native_async_promise_has_token(promise)); + assert!(!pinned(promise)); + assert_eq!(unsafe { (*promise).native_pinned }, 0); +} + +#[test] +fn classify_native_promise_addr_names_null_live_and_reused_slots() { + let _guard = test_native_async_lock(); + assert_eq!(classify_native_promise_addr(0), NativePromiseAddr::Null); + let promise = js_promise_new_cross_thread(); + assert_eq!( + classify_native_promise_addr(promise as usize), + NativePromiseAddr::Live(promise) + ); + // A malloc-resident object of another type where a promise used to be. + let occupant = crate::gc::gc_malloc(64, crate::gc::GC_TYPE_STRING) as usize; + assert_eq!( + classify_native_promise_addr(occupant), + NativePromiseAddr::WrongType(crate::gc::GC_TYPE_STRING) + ); + // Not a heap object at all. + assert_eq!( + classify_native_promise_addr(0x10), + NativePromiseAddr::NotAHeapObject + ); + js_promise_resolve(promise, 0.0); +} diff --git a/crates/perry-runtime/src/promise/mod.rs b/crates/perry-runtime/src/promise/mod.rs index 8635855fe5..51254b37d3 100644 --- a/crates/perry-runtime/src/promise/mod.rs +++ b/crates/perry-runtime/src/promise/mod.rs @@ -22,6 +22,8 @@ pub mod assimilate; pub mod async_step; pub mod checked_dispatch; pub mod combinators; +#[cfg(test)] +mod cross_thread_pin_tests; pub(crate) mod keyed_table; pub mod microtasks; pub mod native_async; @@ -532,6 +534,12 @@ pub type ClosurePtr = *const crate::closure::ClosureHeader; pub struct Promise { /// Current state of the promise pub(crate) state: PromiseState, + /// #9552 — non-zero while this promise holds the cross-thread pin taken by + /// `js_promise_new_cross_thread`. Lives in the padding after `state`, so no + /// other field moves. Cleared, and the pin released, by the settlement + /// paths (`js_promise_resolve` / `js_promise_reject`) and by + /// `remove_token_from_registry` for a token dropped without settling. + pub(crate) native_pinned: u8, /// The resolved value (if fulfilled) pub(crate) value: f64, /// The rejection reason (if rejected) @@ -565,6 +573,7 @@ impl Promise { pub(crate) fn new() -> Self { Promise { state: PromiseState::Pending, + native_pinned: 0, value: 0.0, reason: 0.0, on_fulfilled: ptr::null(), @@ -1264,3 +1273,66 @@ pub extern "C" fn js_microtasks_pending() -> i32 { } TASK_QUEUE.with(|q| if q.borrow().is_empty() { 0 } else { 1 }) } + +/// #9552 — what a raw promise address handed back by native code names. +#[derive(Debug, PartialEq, Eq)] +pub enum NativePromiseAddr { + /// A null hand-off (a caller that never minted a promise). + Null, + /// A live promise. + Live(*mut Promise), + /// Not a tracked heap object at all (freed and unmapped, or never one). + NotAHeapObject, + /// A heap object of another type: the promise was freed and its slot + /// reused. The payload is the occupant's `obj_type`. + WrongType(u8), +} + +/// Classify `addr` without dereferencing anything the heap does not vouch +/// for. Pure, so the abort policy in [`native_promise_from_raw`] is testable. +pub fn classify_native_promise_addr(addr: usize) -> NativePromiseAddr { + if addr == 0 { + return NativePromiseAddr::Null; + } + match unsafe { crate::value::addr_class::try_read_gc_header(addr) } { + None => NativePromiseAddr::NotAHeapObject, + Some(header) if header.obj_type == crate::gc::GC_TYPE_PROMISE => { + NativePromiseAddr::Live(addr as *mut Promise) + } + Some(header) => NativePromiseAddr::WrongType(header.obj_type), + } +} + +/// The trust boundary for a promise address that left the runtime as a bare +/// `usize` (a worker future, a pending-result queue, a native async token) and +/// is now coming back to be settled (#9552). +/// +/// A stale address here is a use-after-free in the making: `js_promise_resolve` +/// would write a state byte and a value into whatever the allocator has since +/// put in the slot, and the corruption surfaces cycles later in an unrelated +/// object (the #9552 report was a RegExp header read as a promise's `next`). +/// Aborting at the boundary names the site and the occupant instead. This runs +/// once per native completion — never per `await` — so it is not on any hot +/// path. +pub fn native_promise_from_raw(addr: usize, site: &str) -> *mut Promise { + match classify_native_promise_addr(addr) { + NativePromiseAddr::Null => ptr::null_mut(), + NativePromiseAddr::Live(promise) => promise, + NativePromiseAddr::NotAHeapObject => { + eprintln!( + "[perry] FATAL (#9552): {site} handed back promise address {addr:#x}, which is \ + not a tracked heap object — the promise was freed while native code still \ + held its address. It was not rooted across its in-flight window." + ); + std::process::abort() + } + NativePromiseAddr::WrongType(obj_type) => { + eprintln!( + "[perry] FATAL (#9552): {site} handed back promise address {addr:#x}, but the \ + object there now has obj_type={obj_type} — the promise was freed while native \ + code still held its address and the slot was reused." + ); + std::process::abort() + } + } +} diff --git a/crates/perry-runtime/src/promise/native_async.rs b/crates/perry-runtime/src/promise/native_async.rs index 0801037cce..98264fbc76 100644 --- a/crates/perry-runtime/src/promise/native_async.rs +++ b/crates/perry-runtime/src/promise/native_async.rs @@ -231,6 +231,12 @@ fn payload_to_settlement(payload: PendingPayload) -> (bool, u64, u32) { } fn remove_token_from_registry(token_ptr: usize, promise: usize) { + // #9552: the registry entry was the token promise's root; the constructor + // pin must not outlive it, or a cancelled token (no settlement, so no + // `js_promise_resolve` to release it) would leak its promise forever. + if promise != 0 { + unsafe { super::then::release_native_pin(promise as *mut Promise) }; + } let mut registry = crate::gc::lock_gc_root_registry(registry()); registry.tokens.retain(|&candidate| candidate != token_ptr); registry.pending.retain(|&candidate| candidate != token_ptr); @@ -506,7 +512,12 @@ pub extern "C" fn js_native_async_process_pending() -> i32 { }; let scope = crate::gc::RuntimeHandleScope::new(); - let promise_handle = scope.root_raw_mut_ptr(promise as *mut Promise); + // #9552: the token carried the address as a bare usize; verify it + // still names a promise before rooting and settling it. + let promise_handle = scope.root_raw_mut_ptr(super::native_promise_from_raw( + promise, + "native async token pump", + )); let handle_roots: Vec<_> = handles .iter() .map(|handle| scope.root_nanbox_u64(handle.value_bits)) diff --git a/crates/perry-runtime/src/promise/then.rs b/crates/perry-runtime/src/promise/then.rs index c2590921d9..c1a550b085 100644 --- a/crates/perry-runtime/src/promise/then.rs +++ b/crates/perry-runtime/src/promise/then.rs @@ -53,13 +53,33 @@ pub(crate) fn js_promise_new_with_parent(parent: *mut Promise) -> *mut Promise { } /// Allocate a Promise that will cross a thread boundary as a raw address -/// (`spawn`, `Atomics.waitAsync`): pinned by the caller and referenced only -/// by a `usize` in the global PENDING_THREAD_RESULTS queue, which no root -/// scanner visits. A nursery resident in that situation is destroyed by the -/// copied-minor from-space flip regardless of its PIN flag (the flip resets -/// eden/survivor blocks wholesale; only root-reachable pins force the -/// fallback). Malloc space is non-moving and both sweep paths honor -/// GC_FLAG_PINNED, so these promises are allocated there unconditionally. +/// (`spawn`, `Atomics.waitAsync`, every stdlib `fetch`/db/ws request that +/// settles through the stdlib pump): referenced only by a `usize` inside a +/// worker future or a pending-result queue, which no root scanner visits. +/// +/// Two properties follow, and this constructor owns both (#9552): +/// +/// * **Non-moving.** A nursery resident is destroyed by the copied-minor +/// from-space flip regardless of its PIN flag (the flip resets +/// eden/survivor blocks wholesale; only root-reachable pins force the +/// fallback). Malloc space is non-moving, so these promises are +/// allocated there unconditionally. +/// * **Rooted until it settles.** Nothing on the JS side points AT a +/// pending promise whose only consumer is an `await` continuation — +/// `P.on_fulfilled = step` and `P.next = N` are edges OUT of `P`, and the +/// worker's `usize` is invisible to the collector. Both sweep paths honor +/// `GC_FLAG_PINNED`, so the constructor pins here and the settlement +/// paths (`js_promise_resolve` / `js_promise_reject`) release the pin the +/// moment the native side is done with the address. Before this lived +/// here, the pin was the CALLER's job, and ~110 stdlib call sites +/// (`fetch` among them) never took it: a full collection landing while a +/// request was in flight freed the promise, and the completion then +/// resolved whatever the allocator had put in its place. +/// +/// The pin is one flag bit on an object this function is already writing, +/// and the release is one byte test on the settlement path; neither touches +/// the young-pin latch that pessimises copying minors (malloc residents are +/// never young-arena). #[no_mangle] pub extern "C" fn js_promise_new_cross_thread() -> *mut Promise { js_promise_new_with_parent_impl(ptr::null_mut(), true) @@ -89,6 +109,14 @@ fn js_promise_new_with_parent_impl(parent: *mut Promise, force_malloc: bool) -> unsafe { // GC_STORE_AUDIT(INIT): initializes freshly allocated Promise storage before the promise is published. ptr::write(promise, Promise::new()); + if force_malloc { + // #9552: see `js_promise_new_cross_thread`. The object is + // malloc-resident (never young-arena), so the non-young pin is + // the right one — it must not arm the copying minor's young-pin + // latch. + crate::gc::pin_user_ptr_non_young(promise as *mut u8); + (*promise).native_pinned = 1; + } let trigger_async_id = parent_handle.with_mut_ptr::(|parent| { if parent.is_null() { crate::async_hooks::execution_async_id_u64() @@ -203,6 +231,20 @@ pub extern "C" fn js_promise_result(promise: *mut Promise) -> f64 { } /// Resolve a promise with a value +/// #9552 — release the cross-thread pin `js_promise_new_cross_thread` took, +/// if this promise holds one. Called from every state transition out of +/// `Pending` and from the token registry when a token is dropped without a +/// settlement. Idempotent: the byte is cleared with the pin, so a second call +/// is one load and a not-taken branch. Ordinary (arena) promises pay exactly +/// that load; the byte shares `state`'s cache line. +#[inline] +pub(crate) unsafe fn release_native_pin(promise: *mut Promise) { + if (*promise).native_pinned != 0 { + (*promise).native_pinned = 0; + crate::gc::unpin_user_ptr(promise as *mut u8); + } +} + #[no_mangle] pub extern "C" fn js_promise_resolve(promise: *mut Promise, value: f64) { if promise.is_null() { @@ -214,6 +256,7 @@ pub extern "C" fn js_promise_resolve(promise: *mut Promise, value: f64) { } super::async_step::trace_async_settle(promise, "fulfill"); (*promise).state = PromiseState::Fulfilled; + release_native_pin(promise); store_promise_jsvalue_slot(promise, std::ptr::addr_of_mut!((*promise).value), value); crate::async_hooks::promise_resolve((*promise).async_id); crate::v8::promise_hook_settled(promise); @@ -422,6 +465,7 @@ pub extern "C" fn js_promise_reject(promise: *mut Promise, reason: f64) { } super::async_step::trace_async_settle(promise, "reject"); (*promise).state = PromiseState::Rejected; + release_native_pin(promise); store_promise_jsvalue_slot(promise, std::ptr::addr_of_mut!((*promise).reason), reason); crate::async_hooks::promise_resolve((*promise).async_id); crate::v8::promise_hook_settled(promise); diff --git a/crates/perry-runtime/src/thread.rs b/crates/perry-runtime/src/thread.rs index d33dfbd224..389cf4c2a6 100644 --- a/crates/perry-runtime/src/thread.rs +++ b/crates/perry-runtime/src/thread.rs @@ -1541,13 +1541,10 @@ unsafe fn spawn_impl(closure_val: f64) -> *mut crate::promise::Promise { // in PENDING_THREAD_RESULTS (no scanner) until drain — a nursery // resident would be destroyed by the copied-minor from-space flip even // while pinned. Malloc space is non-moving and sweeps honor the pin. + // #9552: the cross-thread constructor pins the promise; the settlement in + // `js_thread_process_pending` releases it. let promise = crate::promise::js_promise_new_cross_thread(); - // Pin the promise so GC doesn't collect it while the thread is running. - // Malloc-resident (see above), so this does not arm the young-pin latch. - let promise_header = (promise as *mut u8).sub(gc::GC_HEADER_SIZE) as *mut gc::GcHeader; - gc::pin_object_non_young(promise_header); - let promise_usize = promise as usize; // #6185: the promise lives in the SPAWNING agent's heap, so that is the // agent allowed to settle it. Captured here, on the spawning thread — @@ -1681,16 +1678,6 @@ pub fn thread_job_begin() { ACTIVE_THREAD_JOBS.fetch_add(1, Ordering::SeqCst); } -/// Pin `promise` so GC keeps it alive while a background job runs; the matching -/// unpin happens in [`js_thread_process_pending`] when the result resolves. -/// -/// # Safety -/// `promise` must be a live promise allocation preceded by an 8-byte GcHeader. -pub unsafe fn pin_promise(promise: *mut crate::promise::Promise) { - let header = (promise as *mut u8).sub(gc::GC_HEADER_SIZE) as *mut gc::GcHeader; - gc::pin_object_non_young(header); -} - /// Resolve the promise at `promise_usize` with a UTF-8 string on the agent that /// owns it. Routes through the same pending-result path `spawn` uses (which /// unpins the promise, deserializes the value into that agent's arena, @@ -1787,11 +1774,13 @@ pub extern "C" fn js_thread_process_pending() -> i32 { // `queue_thread_result` (deadlock on a re-entrant lock of the same Mutex). for item in mine { unsafe { - let promise = item.promise_ptr as *mut crate::promise::Promise; - - // Unpin the promise now that we're settling it. - let promise_header = (promise as *mut u8).sub(gc::GC_HEADER_SIZE) as *mut gc::GcHeader; - gc::unpin_object(promise_header); + // #9552: the address crossed the thread boundary as a bare usize; + // verify it still names a promise. The constructor's pin is + // released by the settlement below, not here. + let promise = crate::promise::native_promise_from_raw( + item.promise_ptr, + "perry/thread result drain", + ); // #6185: a worker that returned a non-transferable value (e.g. // `spawn(() => new Map())`) can't throw on its own thread (no diff --git a/crates/perry-stdlib/src/common/async_bridge.rs b/crates/perry-stdlib/src/common/async_bridge.rs index 86e1e42157..793a640b0a 100644 --- a/crates/perry-stdlib/src/common/async_bridge.rs +++ b/crates/perry-stdlib/src/common/async_bridge.rs @@ -87,34 +87,22 @@ fn release_native_async_token(promise: *mut perry_runtime::Promise) { perry_runtime::promise::js_native_async_drop_promise_token(promise); } -/// Allocate a fresh Promise and pin it for cross-thread resolution. -/// Convenience wrapper for direct callers of [`queue_promise_resolution`] -/// / [`queue_deferred_resolution`] (fetch, zlib, bcrypt, ioredis, ws, -/// etc.) — modules that bypass `spawn_for_promise[_deferred]` because -/// their own future setup is custom. Equivalent to -/// `js_promise_new()` followed by [`pin_promise_for_native_resolution`]. +/// Allocate a fresh Promise for cross-thread resolution. Convenience wrapper +/// for direct callers of [`queue_promise_resolution`] / +/// [`queue_deferred_resolution`] — modules that bypass +/// `spawn_for_promise[_deferred]` because their own future setup is custom. +/// +/// #9552: the pin is taken by `js_promise_new_cross_thread` itself and +/// released when the promise settles, so this is now exactly that +/// constructor. Callers that reach for the bare constructor get the same +/// guarantee; this name survives for the modules that spell the intent. /// /// # Safety -/// Same as `js_promise_new()`; the pinning has no preconditions of -/// its own. The matching unpin runs automatically in -/// `js_stdlib_process_pending`. +/// Same as `js_promise_new()`. #[inline] pub unsafe fn js_promise_new_for_native_resolution() -> *mut perry_runtime::Promise { ensure_gc_scanner_registered(); - // #8770: allocate in MALLOC space (non-moving), not the nursery arena. A - // native-resolution promise is handed to a tokio worker as a raw `usize` and, - // until its resolution is queued into PENDING_RESOLUTIONS (which the root - // scanner visits), it is reachable only through that worker-thread capture — - // invisible to the main-thread copying minor. A nursery resident in that - // window is wiped by the from-space flip REGARDLESS of its PIN flag (the flip - // resets eden/survivor blocks wholesale; only root-reachable pins force the - // fallback — see `js_promise_new_cross_thread`). Then `js_stdlib_process_ - // pending` unpins/resolves through the stale pointer and faults on the - // reclaimed header. Malloc space is non-moving and both sweep paths honor - // GC_FLAG_PINNED, so the pin actually protects it there. - let p = perry_runtime::js_promise_new_cross_thread(); - pin_promise_for_native_resolution(p as usize); - p + perry_runtime::js_promise_new_cross_thread() } /// Count of in-flight `perry_ffi_spawn_blocking[_with_reactor]` tasks @@ -550,8 +538,11 @@ pub extern "C" fn js_stdlib_process_pending() -> i32 { for resolution in simple_resolutions { let scope = perry_runtime::gc::RuntimeHandleScope::new(); let promise_ptr_usize = resolution.promise_ptr; - let promise_handle = - scope.root_raw_mut_ptr(promise_ptr_usize as *mut perry_runtime::Promise); + // #9552: the address spent its in-flight window as a bare usize in a + // worker future; verify it still names a promise before touching it. + let promise_handle = scope.root_raw_mut_ptr( + perry_runtime::promise::native_promise_from_raw(promise_ptr_usize, "stdlib pump"), + ); let result_handle = scope.root_nanbox_u64(resolution.result_bits); // Issue #859: unpin BEFORE resolve so the just-settled promise // can be reclaimed by the next GC. Resolve doesn't trigger GC @@ -584,8 +575,11 @@ pub extern "C" fn js_stdlib_process_pending() -> i32 { for resolution in deferred_resolutions { let scope = perry_runtime::gc::RuntimeHandleScope::new(); let promise_ptr_usize = resolution.promise_ptr; - let promise_handle = - scope.root_raw_mut_ptr(promise_ptr_usize as *mut perry_runtime::Promise); + // #9552: the address spent its in-flight window as a bare usize in a + // worker future; verify it still names a promise before touching it. + let promise_handle = scope.root_raw_mut_ptr( + perry_runtime::promise::native_promise_from_raw(promise_ptr_usize, "stdlib pump"), + ); // Run the converter on the main thread to create JSValues safely let result_bits = (resolution.converter)(); let result_handle = scope.root_nanbox_u64(result_bits); @@ -1053,6 +1047,44 @@ where }); } +/// Spawn an async operation whose success and error values both need to be +/// materialized on the main thread. +/// +/// Database adapters use this variant to reject with a real JavaScript Error +/// (including driver-specific fields such as `code` and `errno`) instead of a +/// bare string. As with [`spawn_for_promise_deferred`], neither converter runs +/// on the async executor, where allocating Perry heap values would be unsafe. +/// +/// # Safety +/// `promise_ptr` must point to a live Perry Promise. +pub unsafe fn spawn_for_promise_deferred_with_error( + promise_ptr: *mut u8, + future: F, + converter: C, + reject_converter: R, +) where + T: Send + 'static, + E: Send + 'static, + F: Future> + Send + 'static, + C: FnOnce(T) -> u64 + Send + 'static, + R: FnOnce(E) -> u64 + Send + 'static, +{ + ensure_pump_registered(); + ensure_gc_scanner_registered(); + let ptr = promise_ptr as usize; + pin_promise_for_native_resolution(ptr); + + EXT_BLOCKING_TASKS_INFLIGHT.fetch_add(1, Ordering::AcqRel); + RUNTIME.spawn(async move { + match future.await { + Ok(data) => queue_deferred_resolution(ptr, true, move || converter(data)), + Err(error) => queue_deferred_resolution(ptr, false, move || reject_converter(error)), + } + EXT_BLOCKING_TASKS_INFLIGHT.fetch_sub(1, Ordering::AcqRel); + perry_runtime::event_pump::js_notify_main_thread(); + }); +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/perry-stdlib/src/common/dispatch/method_dispatch.rs b/crates/perry-stdlib/src/common/dispatch/method_dispatch.rs index 53d7ae7ef8..24697034e1 100644 --- a/crates/perry-stdlib/src/common/dispatch/method_dispatch.rs +++ b/crates/perry-stdlib/src/common/dispatch/method_dispatch.rs @@ -1,4 +1,8 @@ -#[cfg(any(feature = "crypto", feature = "database-redis"))] +#[cfg(any( + feature = "crypto", + feature = "database-redis", + feature = "bundled-mysql2" +))] use super::super::handle::with_handle; use super::*; @@ -211,6 +215,13 @@ pub unsafe extern "C" fn js_handle_method_dispatch( return value; } + // mysql2 handles frequently pass through interface-typed fields in Drizzle, + // which removes the static class information used by native lowering. + #[cfg(feature = "bundled-mysql2")] + if let Some(value) = crate::mysql2::dispatch_mysql2_method(handle, method_name, &args) { + return value; + } + // node:sqlite DatabaseSync handle. Keep this before the better-sqlite3 // SQLite fallbacks because method names like prepare/exec/close overlap // but the lifecycle/error semantics are intentionally different. diff --git a/crates/perry-stdlib/src/common/dispatch/property_dispatch.rs b/crates/perry-stdlib/src/common/dispatch/property_dispatch.rs index 77011a0df7..f698902cdc 100644 --- a/crates/perry-stdlib/src/common/dispatch/property_dispatch.rs +++ b/crates/perry-stdlib/src/common/dispatch/property_dispatch.rs @@ -1,4 +1,8 @@ -#[cfg(any(feature = "crypto", feature = "http-client"))] +#[cfg(any( + feature = "crypto", + feature = "http-client", + feature = "bundled-mysql2" +))] use super::super::handle::with_handle; use super::*; @@ -43,6 +47,11 @@ pub unsafe extern "C" fn js_handle_property_dispatch( return value; } + #[cfg(feature = "bundled-mysql2")] + if let Some(value) = crate::mysql2::dispatch_mysql2_property(handle, property_name) { + return value; + } + #[cfg(all( feature = "tls-runtime", not(target_os = "ios"), diff --git a/crates/perry-stdlib/src/mysql2/connection.rs b/crates/perry-stdlib/src/mysql2/connection.rs index 9670cab22d..b2d3d56fb8 100644 --- a/crates/perry-stdlib/src/mysql2/connection.rs +++ b/crates/perry-stdlib/src/mysql2/connection.rs @@ -1,535 +1,245 @@ -//! MySQL connection implementation +//! MySQL connection implementation. +use std::sync::Arc; use std::time::Duration; use perry_runtime::{js_promise_new_cross_thread, JSValue, Promise}; use sqlx::mysql::MySqlConnection; use sqlx::Connection; +use tokio::sync::Mutex; -use super::pool::{extract_params_from_jsvalue, MysqlPoolConnectionHandle, ParamValue}; -use super::result::{is_row_returning_query, QueryOutcome, RawQueryResult}; +use super::pool::{ + execute_query_on_connection, parse_query_request, MysqlPoolConnectionHandle, MysqlPromiseError, + QueryRequest, DEFAULT_QUERY_TIMEOUT_SECS, +}; +use super::result::QueryOutcome; use super::types::parse_mysql_config; -use crate::common::{get_handle_mut, register_handle, Handle}; +use crate::common::{register_handle, take_handle, with_handle, Handle}; -/// Default timeout for connecting to the database (in seconds) -const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 10; -/// Default timeout for overall query operation (in seconds) -const DEFAULT_QUERY_TIMEOUT_SECS: u64 = 30; +const CONNECT_TIMEOUT_SECS: u64 = 10; -/// Wrapper around MySqlConnection that we can store in the handle registry 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))), } } - - pub fn take(&mut self) -> Option { - self.connection.take() - } } -/// mysql.createConnection(config) -> Promise -/// -/// Creates a new MySQL connection with the given configuration. -/// Returns a Promise that resolves to a connection handle. -/// -/// # Safety -/// The config parameter must be a valid JSValue representing a config object. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_create_connection(config_f: f64) -> *mut Promise { - // Take f64 at the FFI boundary to avoid SysV AMD64 ABI mismatch - // (see js_mysql2_create_pool for details). - let config = JSValue::from_bits(config_f.to_bits()); - let promise = js_promise_new_cross_thread(); - - // Parse the config - let mysql_config = parse_mysql_config(config); - - crate::common::spawn_for_promise(promise as *mut u8, async move { - use tokio::time::timeout; - - let url = mysql_config.to_url(); - - // Wrap connection in a timeout to prevent indefinite hangs - match timeout( - Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS), - MySqlConnection::connect(&url), - ) - .await - { - Ok(Ok(conn)) => { - let handle = register_handle(MysqlConnectionHandle::new(conn)); - // NaN-box the handle with POINTER_TAG so it can be properly extracted later - let nanboxed = perry_runtime::js_nanbox_pointer(handle as i64); - Ok(nanboxed.to_bits()) - } - Ok(Err(e)) => Err(format!("Failed to connect: {}", e)), - Err(_) => Err(format!( - "Connection timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_CONNECT_TIMEOUT_SECS - )), - } - }); - - promise +#[derive(Clone)] +pub(crate) enum MysqlConnectionTarget { + Direct(Arc>>), + Pool(Arc>>>), } -/// connection.end() -> Promise -/// -/// Closes the MySQL connection. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_connection_end(conn_handle: Handle) -> *mut Promise { - let promise = js_promise_new_cross_thread(); - - crate::common::spawn_for_promise(promise as *mut u8, async move { - use crate::common::take_handle; - use tokio::time::timeout; +pub(crate) 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)) + }) + }) +} - if let Some(mut wrapper) = take_handle::(conn_handle) { - if let Some(conn) = wrapper.take() { - match timeout( - Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS), - conn.close(), - ) - .await - { - Ok(Ok(())) => Ok(JSValue::undefined().bits()), - Ok(Err(e)) => Err(format!("Failed to close connection: {}", e)), - Err(_) => { - // Connection close timed out, but we've already taken it so just return - Ok(JSValue::undefined().bits()) - } - } - } else { - Err("Connection already closed".to_string()) - } - } else { - Err("Invalid connection handle".to_string()) +async fn execute_query_on_target( + target: MysqlConnectionTarget, + request: &QueryRequest, +) -> Result { + match target { + MysqlConnectionTarget::Direct(connection) => { + let mut slot = connection.lock().await; + let connection = slot + .as_mut() + .ok_or_else(|| MysqlPromiseError::message("Connection already closed"))?; + execute_query_on_connection(connection, request).await } - }); - - promise + MysqlConnectionTarget::Pool(connection) => { + let mut slot = connection.lock().await; + let connection = slot + .as_mut() + .ok_or_else(|| MysqlPromiseError::message("Pool connection released"))?; + execute_query_on_connection(connection, request).await + } + } } -/// connection.query(sql, params?) -> Promise<[rows, fields]> -/// -/// Executes a query and returns the results. -/// This function handles both regular connections (MysqlConnectionHandle) -/// and pool connections (MysqlPoolConnectionHandle). See -/// `js_mysql2_pool_query` for rationale on accepting and binding `params` -/// here. Issue #414. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_connection_query( +unsafe fn run_connection_query( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, + force_prepared: bool, ) -> *mut Promise { let promise = js_promise_new_cross_thread(); - let params = JSValue::from_bits(params_f.to_bits()); - - // Extract the SQL string - let sql = if sql_ptr.is_null() { - String::new() - } else { - let header = sql_ptr as *const perry_runtime::StringHeader; - let len = (*header).byte_len as usize; - let data_ptr = sql_ptr.add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - String::from_utf8_lossy(bytes).to_string() - }; - - let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); - - // Use spawn_for_promise_deferred to safely create JSValues on the main thread - crate::common::spawn_for_promise_deferred( + let request = parse_query_request(query_f, params_f, force_prepared); + let target = connection_target(conn_handle); + let rows_as_array = request + .as_ref() + .map(|request| request.rows_as_array) + .unwrap_or(false); + + crate::common::spawn_for_promise_deferred_with_error( promise as *mut u8, async move { - use tokio::time::timeout; + let request = request?; + let target = + target.ok_or_else(|| MysqlPromiseError::message("Invalid connection handle"))?; + execute_query_on_target(target, &request).await + }, + move |outcome| outcome.to_jsvalue_with_rows_as_array(rows_as_array).bits(), + MysqlPromiseError::to_jsvalue_bits, + ); + promise +} - let param_values = param_values?; +pub(crate) fn transaction_sql_for_method(method: &str) -> Option<&'static str> { + match method { + "beginTransaction" => Some("START TRANSACTION"), + "commit" => Some("COMMIT"), + "rollback" => Some("ROLLBACK"), + _ => None, + } +} - // First try as a regular connection - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(conn) = wrapper.connection.as_mut() { - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } - if is_select { - let query_future = query.fetch_all(conn); - match timeout(Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), query_future).await { - Ok(Ok(rows)) => { - let raw_result = RawQueryResult::from_mysql_rows(rows); - return Ok(QueryOutcome::Rows(raw_result)); - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => return Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), +pub(crate) fn run_simple_command(conn_handle: Handle, sql: &'static str) -> *mut Promise { + let promise = js_promise_new_cross_thread(); + let target = connection_target(conn_handle); + unsafe { + crate::common::spawn_for_promise_deferred_with_error( + promise as *mut u8, + async move { + let target = target + .ok_or_else(|| MysqlPromiseError::message("Invalid connection handle"))?; + let execute = async { + match target { + MysqlConnectionTarget::Direct(connection) => { + let mut slot = connection.lock().await; + let connection = slot.as_mut().ok_or_else(|| { + MysqlPromiseError::message("Connection already closed") + })?; + sqlx::raw_sql(sql) + .execute(connection) + .await + .map_err(|error| MysqlPromiseError::from_sqlx(sql, error))?; } - } else { - let query_future = query.execute(conn); - match timeout(Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), query_future).await { - Ok(Ok(result)) => { - return Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }); - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => return Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), + MysqlConnectionTarget::Pool(connection) => { + let mut slot = connection.lock().await; + let connection = slot.as_mut().ok_or_else(|| { + MysqlPromiseError::message("Pool connection released") + })?; + sqlx::raw_sql(sql) + .execute(&mut **connection) + .await + .map_err(|error| MysqlPromiseError::from_sqlx(sql, error))?; } } - } else { - return Err("Connection already closed".to_string()); - } - } + Ok::<_, MysqlPromiseError>(JSValue::undefined().bits()) + }; + tokio::time::timeout(Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), execute) + .await + .map_err(|_| MysqlPromiseError::message(format!("{sql} timed out")))? + }, + |bits| bits, + MysqlPromiseError::to_jsvalue_bits, + ); + } + promise +} - // Then try as a pool connection - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(ref mut conn) = wrapper.connection { - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } - if is_select { - let query_future = query.fetch_all(&mut **conn); - match timeout(Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), query_future).await { - Ok(Ok(rows)) => { - let raw_result = RawQueryResult::from_mysql_rows(rows); - return Ok(QueryOutcome::Rows(raw_result)); - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => return Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - let query_future = query.execute(&mut **conn); - match timeout(Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), query_future).await { - Ok(Ok(result)) => { - return Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }); - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => return Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } - } else { - return Err("Connection has been released".to_string()); - } - } +/// mysql.createConnection(config) -> Promise. +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_create_connection(config_f: f64) -> *mut Promise { + let config = JSValue::from_bits(config_f.to_bits()); + let mysql_config = parse_mysql_config(config); + let promise = js_promise_new_cross_thread(); - Err("Invalid connection handle".to_string()) + crate::common::spawn_for_promise_deferred_with_error( + promise as *mut u8, + async move { + let connection = tokio::time::timeout( + Duration::from_secs(CONNECT_TIMEOUT_SECS), + MySqlConnection::connect(&mysql_config.to_url()), + ) + .await + .map_err(|_| MysqlPromiseError::message("MySQL connection timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Failed to connect", error))?; + Ok(connection) }, - |outcome: QueryOutcome| outcome.to_jsvalue().bits(), + |connection| { + let handle = register_handle(MysqlConnectionHandle::new(connection)); + perry_runtime::js_nanbox_pointer(handle).to_bits() + }, + MysqlPromiseError::to_jsvalue_bits, ); - promise } -/// connection.execute(sql, params) -> Promise<[rows, fields]> -/// -/// Executes a prepared statement with parameters. #[no_mangle] -pub unsafe extern "C" fn js_mysql2_connection_execute( - conn_handle: Handle, - sql_ptr: *const u8, - params_f: f64, -) -> *mut Promise { +pub unsafe extern "C" fn js_mysql2_connection_end(conn_handle: Handle) -> *mut Promise { let promise = js_promise_new_cross_thread(); - let params = JSValue::from_bits(params_f.to_bits()); - - let sql = if sql_ptr.is_null() { - String::new() - } else { - let header = sql_ptr as *const perry_runtime::StringHeader; - let len = (*header).byte_len as usize; - let data_ptr = sql_ptr.add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - String::from_utf8_lossy(bytes).to_string() - }; - - let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); - - crate::common::spawn_for_promise_deferred( + let connection = take_handle::(conn_handle) + .map(|wrapper| Arc::clone(&wrapper.connection)); + crate::common::spawn_for_promise_deferred_with_error( promise as *mut u8, async move { - use tokio::time::timeout; - - let param_values = param_values?; - - // Try as a regular connection first - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(conn) = wrapper.connection.as_mut() { - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } - if is_select { - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query.fetch_all(conn), - ) - .await - { - Ok(Ok(rows)) => { - return Ok(QueryOutcome::Rows(RawQueryResult::from_mysql_rows( - rows, - ))) - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => { - return Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )) - } - } - } else { - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query.execute(conn), - ) - .await - { - Ok(Ok(result)) => { - return Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }) - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => { - return Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )) - } - } - } - } else { - return Err("Connection already closed".to_string()); - } - } - - // Try as a pool connection - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(ref mut conn) = wrapper.connection { - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } - if is_select { - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query.fetch_all(&mut **conn), - ) - .await - { - Ok(Ok(rows)) => { - return Ok(QueryOutcome::Rows(RawQueryResult::from_mysql_rows( - rows, - ))) - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => { - return Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )) - } - } - } else { - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query.execute(&mut **conn), - ) - .await - { - Ok(Ok(result)) => { - return Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }) - } - Ok(Err(e)) => return Err(format!("Query failed: {}", e)), - Err(_) => { - return Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )) - } - } - } - } else { - return Err("Connection has been released".to_string()); - } - } - - Err("Invalid connection handle".to_string()) + let connection = connection + .ok_or_else(|| MysqlPromiseError::message("Invalid connection handle"))?; + let connection = connection + .lock() + .await + .take() + .ok_or_else(|| MysqlPromiseError::message("Connection already closed"))?; + tokio::time::timeout( + Duration::from_secs(CONNECT_TIMEOUT_SECS), + connection.close(), + ) + .await + .map_err(|_| MysqlPromiseError::message("Connection close timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Failed to close", error))?; + Ok(JSValue::undefined().bits()) }, - |outcome: QueryOutcome| outcome.to_jsvalue().bits(), + |bits| bits, + MysqlPromiseError::to_jsvalue_bits, ); - promise } -/// connection.beginTransaction() -> Promise #[no_mangle] -pub unsafe extern "C" fn js_mysql2_connection_begin_transaction( +pub unsafe extern "C" fn js_mysql2_connection_query( conn_handle: Handle, + query_f: f64, + params_f: f64, ) -> *mut Promise { - let promise = js_promise_new_cross_thread(); - - crate::common::spawn_for_promise(promise as *mut u8, async move { - use crate::common::get_handle_mut; - use tokio::time::timeout; + run_connection_query(conn_handle, query_f, params_f, false) +} - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(conn) = wrapper.connection.as_mut() { - let query_future = sqlx::query("BEGIN").execute(conn); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(_)) => Ok(JSValue::undefined().bits()), - Ok(Err(e)) => Err(format!("Failed to begin transaction: {}", e)), - Err(_) => Err(format!( - "Begin transaction timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - Err("Connection already closed".to_string()) - } - } else { - Err("Invalid connection handle".to_string()) - } - }); +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_connection_execute( + conn_handle: Handle, + query_f: f64, + params_f: f64, +) -> *mut Promise { + run_connection_query(conn_handle, query_f, params_f, true) +} - promise +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_connection_begin_transaction( + conn_handle: Handle, +) -> *mut Promise { + run_simple_command(conn_handle, "START TRANSACTION") } -/// connection.commit() -> Promise #[no_mangle] pub unsafe extern "C" fn js_mysql2_connection_commit(conn_handle: Handle) -> *mut Promise { - let promise = js_promise_new_cross_thread(); - - crate::common::spawn_for_promise(promise as *mut u8, async move { - use crate::common::get_handle_mut; - use tokio::time::timeout; - - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(conn) = wrapper.connection.as_mut() { - let query_future = sqlx::query("COMMIT").execute(conn); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(_)) => Ok(JSValue::undefined().bits()), - Ok(Err(e)) => Err(format!("Failed to commit transaction: {}", e)), - Err(_) => Err(format!( - "Commit timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - Err("Connection already closed".to_string()) - } - } else { - Err("Invalid connection handle".to_string()) - } - }); - - promise + run_simple_command(conn_handle, "COMMIT") } -/// connection.rollback() -> Promise #[no_mangle] pub unsafe extern "C" fn js_mysql2_connection_rollback(conn_handle: Handle) -> *mut Promise { - let promise = js_promise_new_cross_thread(); - - crate::common::spawn_for_promise(promise as *mut u8, async move { - use crate::common::get_handle_mut; - use tokio::time::timeout; - - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(conn) = wrapper.connection.as_mut() { - let query_future = sqlx::query("ROLLBACK").execute(conn); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(_)) => Ok(JSValue::undefined().bits()), - Ok(Err(e)) => Err(format!("Failed to rollback transaction: {}", e)), - Err(_) => Err(format!( - "Rollback timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - Err("Connection already closed".to_string()) - } - } else { - Err("Invalid connection handle".to_string()) - } - }); - - promise + run_simple_command(conn_handle, "ROLLBACK") } diff --git a/crates/perry-stdlib/src/mysql2/mod.rs b/crates/perry-stdlib/src/mysql2/mod.rs index a5878def49..ac9d4f5855 100644 --- a/crates/perry-stdlib/src/mysql2/mod.rs +++ b/crates/perry-stdlib/src/mysql2/mod.rs @@ -11,3 +11,123 @@ pub use connection::*; pub use pool::*; pub use result::*; pub use types::*; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum MysqlHandleKind { + Pool, + PoolConnection, + Connection, +} + +fn mysql_handle_kind(handle: crate::common::Handle) -> Option { + crate::common::with_handle::(handle, |_| MysqlHandleKind::Pool) + .or_else(|| { + crate::common::with_handle::(handle, |_| { + MysqlHandleKind::PoolConnection + }) + }) + .or_else(|| { + crate::common::with_handle::(handle, |_| { + MysqlHandleKind::Connection + }) + }) +} + +fn method_is_available(kind: MysqlHandleKind, method: &str) -> bool { + match kind { + MysqlHandleKind::Pool => { + matches!( + method, + "query" | "execute" | "end" | "getConnection" | "promise" + ) + } + MysqlHandleKind::PoolConnection => matches!( + method, + "query" | "execute" | "release" | "beginTransaction" | "commit" | "rollback" + ), + MysqlHandleKind::Connection => matches!( + method, + "query" | "execute" | "end" | "beginTransaction" | "commit" | "rollback" | "promise" + ), + } +} + +/// Runtime method dispatch for mysql2 handles whose static TypeScript class +/// was erased (notably Drizzle's interface-typed client fields). +pub(crate) unsafe fn dispatch_mysql2_method( + handle: crate::common::Handle, + method: &str, + args: &[f64], +) -> Option { + let kind = mysql_handle_kind(handle)?; + if !method_is_available(kind, method) { + return None; + } + let undefined = crate::common::TAG_UNDEFINED_F64; + let arg = |index: usize| args.get(index).copied().unwrap_or(undefined); + let pointer = |ptr: *mut perry_runtime::Promise| { + f64::from_bits(perry_runtime::JSValue::pointer(ptr as *const u8).bits()) + }; + + Some(match (kind, method) { + (MysqlHandleKind::Pool, "query") => { + pointer(pool::js_mysql2_pool_query(handle, arg(0), arg(1))) + } + (MysqlHandleKind::Pool, "execute") => { + pointer(pool::js_mysql2_pool_execute(handle, arg(0), arg(1))) + } + (MysqlHandleKind::Pool, "getConnection") => { + pointer(pool::js_mysql2_pool_get_connection(handle)) + } + (MysqlHandleKind::Pool, "end") => pointer(pool::js_mysql2_pool_end(handle)), + (MysqlHandleKind::PoolConnection, "query") => pointer( + pool::js_mysql2_pool_connection_query(handle, arg(0), arg(1)), + ), + (MysqlHandleKind::PoolConnection, "execute") => pointer( + pool::js_mysql2_pool_connection_execute(handle, arg(0), arg(1)), + ), + (MysqlHandleKind::PoolConnection, "release") => { + pool::js_mysql2_pool_connection_release(handle); + undefined + } + (MysqlHandleKind::Connection, "query") => pointer(connection::js_mysql2_connection_query( + handle, + arg(0), + arg(1), + )), + (MysqlHandleKind::Connection, "execute") => pointer( + connection::js_mysql2_connection_execute(handle, arg(0), arg(1)), + ), + (MysqlHandleKind::Connection, "end") => { + pointer(connection::js_mysql2_connection_end(handle)) + } + (MysqlHandleKind::PoolConnection | MysqlHandleKind::Connection, method) + if connection::transaction_sql_for_method(method).is_some() => + { + let sql = connection::transaction_sql_for_method(method)?; + pointer(connection::run_simple_command(handle, sql)) + } + (MysqlHandleKind::Pool | MysqlHandleKind::Connection, "promise") => { + crate::common::nanbox_handle_value(handle) + } + _ => return None, + }) +} + +/// Property reads for mysql2 methods return a bound method. This makes +/// `Reflect.has(pool, "getConnection")`, `"getConnection" in pool`, and +/// `typeof pool.getConnection` agree with the real mysql2 objects. +pub(crate) unsafe fn dispatch_mysql2_property( + handle: crate::common::Handle, + property: &str, +) -> Option { + let kind = mysql_handle_kind(handle)?; + if !method_is_available(kind, property) { + return None; + } + Some(perry_runtime::object::js_class_method_bind( + crate::common::nanbox_handle_value(handle), + property.as_ptr(), + property.len(), + )) +} diff --git a/crates/perry-stdlib/src/mysql2/pool.rs b/crates/perry-stdlib/src/mysql2/pool.rs index 80c52f6e41..467de0fdc1 100644 --- a/crates/perry-stdlib/src/mysql2/pool.rs +++ b/crates/perry-stdlib/src/mysql2/pool.rs @@ -1,26 +1,25 @@ -//! MySQL connection pool implementation +//! MySQL connection pool implementation. +use std::sync::Arc; use std::time::Duration; use perry_runtime::{ - js_array_get_jsvalue, js_array_length, js_promise_new_cross_thread, JSValue, Promise, + js_array_get_jsvalue, js_array_length, js_object_get_field_by_name, + js_promise_new_cross_thread, js_string_from_bytes, JSValue, Promise, }; -use sqlx::mysql::{MySqlPool, MySqlPoolOptions}; +use sqlx::mysql::{MySqlConnection, MySqlDatabaseError, MySqlPool, MySqlPoolOptions}; use sqlx::pool::PoolConnection; use sqlx::MySql; +use tokio::sync::Mutex; use super::result::{is_row_returning_query, QueryOutcome, RawQueryResult}; use super::types::parse_mysql_config; -use crate::common::{register_handle, take_handle, Handle}; +use crate::common::{register_handle, take_handle, with_handle, Handle}; -/// Default timeout for acquiring a connection from the pool (in seconds) -const DEFAULT_ACQUIRE_TIMEOUT_SECS: u64 = 10; -/// Default timeout for connecting to the database (in seconds) +pub(crate) const DEFAULT_ACQUIRE_TIMEOUT_SECS: u64 = 10; const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 10; -/// Default timeout for overall query operation (in seconds) -const DEFAULT_QUERY_TIMEOUT_SECS: u64 = 30; +pub(crate) const DEFAULT_QUERY_TIMEOUT_SECS: u64 = 30; -/// Wrapper around MySqlPool pub struct MysqlPoolHandle { pub pool: MySqlPool, } @@ -31,419 +30,360 @@ impl MysqlPoolHandle { } } -/// Wrapper around a pool connection -/// When dropped, the connection is automatically returned to the pool +/// A checked-out pool connection. The registry entry can be removed while an +/// operation is in flight, so the connection itself is shared and serialized. 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))), } } +} - /// Take the connection out of this handle - pub fn take(&mut self) -> Option> { - self.connection.take() - } +#[derive(Clone, Debug, PartialEq)] +pub(crate) enum ParamValue { + Null, + String(String), + Bytes(Vec), + DateTime(chrono::NaiveDateTime), + Number(f64), + Int(i64), + Bool(bool), } -/// mysql.createPool(config) -> Pool -/// -/// Creates a new connection pool. The pool connects lazily, so this -/// returns synchronously. -/// -/// # Safety -/// The config parameter must be a valid JSValue representing a config object. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_create_pool(config_f: f64) -> Handle { - // Take f64 at the FFI boundary to avoid SysV AMD64 ABI mismatch: - // JSValue is `#[repr(transparent)] u64` (integer register), but the - // LLVM call site declares the arg as `double` (XMM register). On ARM64 - // these aliases (d0/x0 same phys reg) so the bug is invisible, but on - // x86_64 they're distinct registers and the pointer bits never arrive. - let config = JSValue::from_bits(config_f.to_bits()); - let mysql_config = parse_mysql_config(config); - let url = mysql_config.to_url(); - - // Create pool with lazy connection using the tokio runtime context - // We need to enter the runtime context for connect_lazy to work - let _guard = crate::common::runtime().enter(); - - // Use eager connection to get immediate error feedback - let pool_result = crate::common::runtime().block_on(async { - MySqlPoolOptions::new() - .max_connections(10) - .acquire_timeout(Duration::from_secs(DEFAULT_ACQUIRE_TIMEOUT_SECS)) - .connect(&url) - .await - }); +/// Owned data for one mysql2 request. No pointer into the Perry heap crosses +/// the async boundary. +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct QueryRequest { + pub(crate) sql: String, + pub(crate) params: Vec, + pub(crate) rows_as_array: bool, + force_prepared: bool, +} - match pool_result { - Ok(pool) => register_handle(MysqlPoolHandle::new(pool)), - Err(_e) => 0, +impl QueryRequest { + 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() } } -/// pool.end() -> Promise -/// -/// Closes all connections in the pool. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_pool_end(pool_handle: Handle) -> *mut Promise { - let promise = js_promise_new_cross_thread(); +#[derive(Debug)] +pub(crate) struct MysqlPromiseError { + message: String, + code: Option<&'static str>, + errno: Option, +} - crate::common::spawn_for_promise(promise as *mut u8, async move { - use crate::common::take_handle; - use tokio::time::timeout; +impl MysqlPromiseError { + pub(crate) fn message(message: impl Into) -> Self { + Self { + message: message.into(), + code: None, + errno: None, + } + } - if let Some(wrapper) = take_handle::(pool_handle) { - // Wrap pool close in a timeout (use shorter timeout since close should be fast) - match timeout( - Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS), - wrapper.pool.close(), - ) - .await - { - Ok(()) => Ok(JSValue::undefined().bits()), - Err(_) => { - // Pool close timed out, but we've already taken the handle so just return - Ok(JSValue::undefined().bits()) - } - } - } else { - Err("Invalid pool handle".to_string()) + pub(crate) fn from_sqlx(context: &str, error: sqlx::Error) -> Self { + let errno = error + .as_database_error() + .and_then(|database| database.try_downcast_ref::()) + .map(MySqlDatabaseError::number); + Self { + message: format!("{context}: {error}"), + code: errno.and_then(mysql2_error_code), + errno, } - }); + } - promise -} + /// Build the rejection value on the main thread. mysql2 rejects with an + /// Error object, not the bare string previously emitted by the fallback. + pub(crate) fn to_jsvalue_bits(self) -> u64 { + if let Some(errno) = self.errno { + let code = self.code.unwrap_or(""); + return unsafe { + perry_runtime::error::js_node_system_error_value( + self.message.as_ptr(), + self.message.len(), + code.as_ptr(), + code.len(), + std::ptr::null(), + 0, + f64::from(errno), + ) + .to_bits() + }; + } -/// pool.query(sql, params?) -> Promise<[rows, fields]> -/// -/// Executes a query using a connection from the pool. -/// -/// `params` is the optional second arg user code passes to `db.query(sql, [..])`. -/// The codegen dispatch table for `("mysql2", "Pool", "query")` declares -/// `args: &[NA_STR, NA_F64]` so the call site always emits 3 arguments -/// (handle + sql + params). When the user omits `params`, codegen pads the -/// slot with JS `undefined`; `extract_params_from_jsvalue` returns an -/// empty Vec for that case. When the user passes an array, sqlx builds a -/// prepared statement and binds each value — same code path as -/// `js_mysql2_pool_execute`. Without binding, sqlx sends the binary execute -/// frame with 1 placeholder but 0 bind values, MySQL replies with error -/// 1835 ("Malformed communication packet"), and the connection becomes -/// unusable. See issue #414. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_pool_query( - pool_handle: Handle, - sql_ptr: *const u8, - params_f: f64, -) -> *mut Promise { - let promise = js_promise_new_cross_thread(); - let params = JSValue::from_bits(params_f.to_bits()); + let message = js_string_from_bytes(self.message.as_ptr(), self.message.len() as u32); + let error = perry_runtime::error::js_error_new_with_message(message); + JSValue::pointer(error as *const u8).bits() + } +} - // Extract the SQL string - let sql = if sql_ptr.is_null() { - String::new() - } else { - let header = sql_ptr as *const perry_runtime::StringHeader; - let len = (*header).byte_len as usize; - let data_ptr = sql_ptr.add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - String::from_utf8_lossy(bytes).to_string() - }; +/// Symbolic names exposed by mysql2 for common server errors. Unknown server +/// errors still carry their numeric `.errno`. +fn mysql2_error_code(errno: u16) -> Option<&'static str> { + Some(match errno { + 1022 => "ER_DUP_KEY", + 1045 => "ER_ACCESS_DENIED_ERROR", + 1048 => "ER_BAD_NULL_ERROR", + 1049 => "ER_BAD_DB_ERROR", + 1050 => "ER_TABLE_EXISTS_ERROR", + 1051 => "ER_BAD_TABLE_ERROR", + 1052 => "ER_NON_UNIQ_ERROR", + 1054 => "ER_BAD_FIELD_ERROR", + 1062 => "ER_DUP_ENTRY", + 1064 => "ER_PARSE_ERROR", + 1146 => "ER_NO_SUCH_TABLE", + 1169 => "ER_DUP_UNIQUE", + 1205 => "ER_LOCK_WAIT_TIMEOUT", + 1213 => "ER_LOCK_DEADLOCK", + 1216 => "ER_NO_REFERENCED_ROW", + 1217 => "ER_ROW_IS_REFERENCED", + 1264 => "ER_WARN_DATA_OUT_OF_RANGE", + 1292 => "ER_TRUNCATED_WRONG_VALUE", + 1364 => "ER_NO_DEFAULT_FOR_FIELD", + 1406 => "ER_DATA_TOO_LONG", + 1451 => "ER_ROW_IS_REFERENCED_2", + 1452 => "ER_NO_REFERENCED_ROW_2", + 1586 => "ER_DUP_ENTRY_WITH_KEY_NAME", + 1830 => "ER_FK_COLUMN_NOT_NULL", + 1834 => "ER_FK_CANNOT_DELETE_PARENT", + 1859 => "ER_DUP_UNKNOWN_IN_INDEX", + 3819 => "ER_CHECK_CONSTRAINT_VIOLATED", + 4025 => "ER_CONSTRAINT_FAILED", + _ => return None, + }) +} - // Extract parameters from the JSValue array (empty Vec when caller - // passed no params — `extract_params_from_jsvalue` short-circuits on - // 0/undefined/non-array). - let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); +unsafe fn jsvalue_to_string(value: JSValue) -> Option { + let mut scratch = [0; perry_runtime::value::SHORT_STRING_MAX_LEN]; + let (ptr, len) = + perry_runtime::string::str_bytes_from_jsvalue(f64::from_bits(value.bits()), &mut scratch)?; + if ptr.is_null() { + return Some(String::new()); + } + let bytes = std::slice::from_raw_parts(ptr, len as usize); + Some(String::from_utf8_lossy(bytes).into_owned()) +} - // Use spawn_for_promise_deferred to safely create JSValues on the main thread - // The async block returns raw Rust data, and the converter creates JSValues - crate::common::spawn_for_promise_deferred( - promise as *mut u8, - async move { - use crate::common::get_handle; - use tokio::time::timeout; - - let param_values = param_values?; - - if let Some(wrapper) = get_handle::(pool_handle) { - // Build the query with parameter bindings (no-op when - // param_values is empty, preserving the no-param call shape). - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } +unsafe fn object_pointer(value: JSValue) -> Option<*const perry_runtime::ObjectHeader> { + if value.is_pointer() { + let ptr = value.as_pointer::(); + return (!ptr.is_null()).then_some(ptr); + } - if is_select { - // SELECT/SHOW/DESCRIBE: fetch rows - let query_future = query.fetch_all(&wrapper.pool); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(rows)) => { - let raw_result = RawQueryResult::from_mysql_rows(rows); - Ok(QueryOutcome::Rows(raw_result)) - } - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - // INSERT/UPDATE/DELETE: execute and return metadata - let query_future = query.execute(&wrapper.pool); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(result)) => Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }), - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } - } else { - Err("Invalid pool handle".to_string()) - } - }, - // Converter runs on main thread - safe to create JSValues here - |outcome: QueryOutcome| outcome.to_jsvalue().bits(), - ); + // Some generic call sites still pass an untagged object pointer. + let bits = value.bits(); + if bits != 0 && bits <= 0x0000_7FFF_FFFF_FFFF { + return Some(bits as *const perry_runtime::ObjectHeader); + } + None +} - promise +unsafe fn object_field(value: JSValue, name: &str) -> JSValue { + // Allocating the lookup key can trigger a moving collection. Root and + // refresh the receiver before dereferencing it afterwards. + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let receiver = scope.root_nanbox_u64(value.bits()); + let key = js_string_from_bytes(name.as_ptr(), name.len() as u32); + let Some(object) = object_pointer(JSValue::from_bits(receiver.get_nanbox_u64())) else { + return JSValue::undefined(); + }; + js_object_get_field_by_name(object, key) } -/// pool.execute(sql, params) -> Promise<[rows, fields]> -/// -/// Executes a prepared statement with parameters using a connection from the pool. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_pool_execute( - pool_handle: Handle, - sql_ptr: *const u8, +/// Parse mysql2's `query(sql, values?)` and `query({ sql, values?, +/// rowsAsArray? }, values?)` forms while all JS values are still rooted by the +/// native call. +pub(crate) unsafe fn parse_query_request( + query_f: f64, params_f: f64, -) -> *mut Promise { - let promise = js_promise_new_cross_thread(); - let params = JSValue::from_bits(params_f.to_bits()); - - // Extract the SQL string - let sql = if sql_ptr.is_null() { - String::new() + force_prepared: bool, +) -> Result { + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let query = scope.root_nanbox_f64(query_f); + let supplied_params = scope.root_nanbox_f64(params_f); + + let query_value = JSValue::from_bits(query.get_nanbox_u64()); + let (sql, rows_as_array, option_values) = if let Some(sql) = jsvalue_to_string(query_value) { + (sql, false, JSValue::undefined()) } else { - let header = sql_ptr as *const perry_runtime::StringHeader; - let len = (*header).byte_len as usize; - let data_ptr = sql_ptr.add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - String::from_utf8_lossy(bytes).to_string() + let sql_value = object_field(JSValue::from_bits(query.get_nanbox_u64()), "sql"); + let sql = jsvalue_to_string(sql_value).ok_or_else(|| { + MysqlPromiseError::message("Query must be a SQL string or an options object with sql") + })?; + let rows_as_array = object_field(JSValue::from_bits(query.get_nanbox_u64()), "rowsAsArray"); + let rows_as_array = rows_as_array.is_bool() && rows_as_array.as_bool(); + ( + sql, + rows_as_array, + object_field(JSValue::from_bits(query.get_nanbox_u64()), "values"), + ) }; - // Extract parameters from the JSValue array - let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); + let supplied_params = JSValue::from_bits(supplied_params.get_nanbox_u64()); + let params = if supplied_params.is_undefined() { + option_values + } else { + supplied_params + }; + let params = extract_params_from_jsvalue(params).map_err(MysqlPromiseError::message)?; + + Ok(QueryRequest { + sql, + params, + rows_as_array, + force_prepared, + }) +} - // Use spawn_for_promise_deferred to safely create JSValues on the main thread - crate::common::spawn_for_promise_deferred( - promise as *mut u8, - async move { - use crate::common::get_handle; - use tokio::time::timeout; - - let param_values = param_values?; - - if let Some(wrapper) = get_handle::(pool_handle) { - // Build the query with parameter bindings - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } +pub(crate) async fn execute_query_on_connection( + conn: &mut MySqlConnection, + request: &QueryRequest, +) -> Result { + let is_select = request.is_row_returning(); + + if !request.uses_prepared_statement() { + // mysql2 `query()` uses MySQL's text protocol when there are no bind + // values. This is required for commands such as BEGIN that the server + // refuses through the prepared-statement protocol (#9517). + let query = sqlx::raw_sql(sqlx::AssertSqlSafe(request.sql.clone())); + if is_select { + let rows = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + query.fetch_all(&mut *conn), + ) + .await + .map_err(|_| MysqlPromiseError::message("Query timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Query failed", error))?; + return Ok(QueryOutcome::Rows(RawQueryResult::from_mysql_rows(rows))); + } - if is_select { - let query_future = query.fetch_all(&wrapper.pool); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(rows)) => { - let raw_result = RawQueryResult::from_mysql_rows(rows); - Ok(QueryOutcome::Rows(raw_result)) - } - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - let query_future = query.execute(&wrapper.pool); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(result)) => Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }), - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds (MySQL server may be unavailable)", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } - } else { - Err("Invalid pool handle".to_string()) - } - }, - |outcome: QueryOutcome| outcome.to_jsvalue().bits(), - ); + let result = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + query.execute(&mut *conn), + ) + .await + .map_err(|_| MysqlPromiseError::message("Query timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Query failed", error))?; + return Ok(QueryOutcome::Executed { + affected_rows: result.rows_affected(), + last_insert_id: result.last_insert_id(), + }); + } - promise -} + // Do not retain prepared statements between calls. This keeps each mysql2 + // request's SQL, bind metadata, and arguments together (#8745). + 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(value) => query.bind(value.clone()), + ParamValue::Bytes(value) => query.bind(value.clone()), + ParamValue::DateTime(value) => query.bind(*value), + ParamValue::Number(value) => query.bind(*value), + ParamValue::Int(value) => query.bind(*value), + ParamValue::Bool(value) => query.bind(*value), + }; + } -/// Enum to hold different parameter value types -#[derive(Clone, Debug, PartialEq)] -pub(crate) enum ParamValue { - Null, - String(String), - Bytes(Vec), - DateTime(chrono::NaiveDateTime), - Number(f64), - Int(i64), - Bool(bool), + if is_select { + let rows = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + query.fetch_all(&mut *conn), + ) + .await + .map_err(|_| MysqlPromiseError::message("Query timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Query failed", error))?; + Ok(QueryOutcome::Rows(RawQueryResult::from_mysql_rows(rows))) + } else { + let result = tokio::time::timeout( + Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), + query.execute(&mut *conn), + ) + .await + .map_err(|_| MysqlPromiseError::message("Query timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Query failed", error))?; + Ok(QueryOutcome::Executed { + affected_rows: result.rows_affected(), + last_insert_id: result.last_insert_id(), + }) + } } -/// Extract parameter values from a JSValue array +/// Extract parameter values from a JS array before scheduling async work. pub(crate) unsafe fn extract_params_from_jsvalue( params: JSValue, ) -> Result, String> { - let mut result = Vec::new(); - - let bits = params.bits(); - - if bits == 0 || params.is_undefined() || params.is_null() { - return Ok(result); + if params.bits() == 0 || params.is_undefined() || params.is_null() { + return Ok(Vec::new()); } - let is_array = - JSValue::from_bits(perry_runtime::js_array_is_array(f64::from_bits(bits)).to_bits()) - .as_bool(); + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let params_handle = scope.root_nanbox_u64(params.bits()); + let is_array = JSValue::from_bits( + perry_runtime::js_array_is_array(params_handle.get_nanbox_f64()).to_bits(), + ) + .as_bool(); if !is_array { return Err("Bind parameters must be an array".to_string()); } - // Handle both NaN-boxed pointers and raw pointers - let arr_ptr: *const perry_runtime::ArrayHeader = if params.is_pointer() { - // NaN-boxed pointer (POINTER_TAG = 0x7FFD) - params.as_pointer() as *const perry_runtime::ArrayHeader + let refreshed_params = JSValue::from_bits(params_handle.get_nanbox_u64()); + let bits = refreshed_params.bits(); + let array: *const perry_runtime::ArrayHeader = if refreshed_params.is_pointer() { + refreshed_params.as_pointer() } else if bits != 0 && bits <= 0x0000_FFFF_FFFF_FFFF { - // Raw pointer (not NaN-boxed) - the bits ARE the pointer - // Check upper bits don't match any NaN-box tag (0x7FFC-0x7FFF) - let upper = bits >> 48; - if upper == 0 || (upper > 0 && upper < 0x7FF0) { - bits as *const perry_runtime::ArrayHeader - } else { - return Err("Bind parameters array has no valid runtime pointer".to_string()); - } + bits as *const perry_runtime::ArrayHeader } else { return Err("Bind parameters array has no valid runtime pointer".to_string()); }; - - if arr_ptr.is_null() { + if array.is_null() { return Err("Bind parameters array has no valid runtime pointer".to_string()); } - let length = js_array_length(arr_ptr); - - for i in 0..length { - let element_bits = js_array_get_jsvalue(arr_ptr, i); + let length = js_array_length(array); + let mut result = Vec::with_capacity(length as usize); + for index in 0..length { + let refreshed_params = JSValue::from_bits(params_handle.get_nanbox_u64()); + let array: *const perry_runtime::ArrayHeader = if refreshed_params.is_pointer() { + refreshed_params.as_pointer() + } else { + refreshed_params.bits() as *const perry_runtime::ArrayHeader + }; + let element_bits = js_array_get_jsvalue(array, index); let element = JSValue::from_bits(element_bits); - - let param = if element.is_null() { + let value = if element.is_null() { ParamValue::Null } else if element.is_undefined() { - return Err(format!("Bind parameter at index {i} is undefined")); - } else if element.is_any_string() { - // Extract string value - if element.is_short_string() { - let mut bytes = [0; perry_runtime::value::SHORT_STRING_MAX_LEN]; - let len = element.short_string_to_buf(&mut bytes); - ParamValue::String(String::from_utf8_lossy(&bytes[..len]).to_string()) - } else { - let str_ptr = element.as_string_ptr(); - if str_ptr.is_null() { - return Err(format!("Could not read string bind parameter at index {i}")); - } - let len = (*str_ptr).byte_len as usize; - let data_ptr = - (str_ptr as *const u8).add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - ParamValue::String(String::from_utf8_lossy(bytes).to_string()) - } + return Err(format!("Bind parameter at index {index} is undefined")); + } else if let Some(value) = jsvalue_to_string(element) { + ParamValue::String(value) } else if element.is_bigint() { - // Convert BigInt to string (MySQL handles numeric strings correctly) - let bigint_ptr = element.as_bigint_ptr(); - if !bigint_ptr.is_null() { - let str_ptr = perry_runtime::bigint::js_bigint_to_string(bigint_ptr); - if !str_ptr.is_null() { - let len = (*str_ptr).byte_len as usize; - let data_ptr = (str_ptr as *const u8) - .add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - ParamValue::String(String::from_utf8_lossy(bytes).to_string()) - } else { - ParamValue::String("0".to_string()) - } - } else { - ParamValue::String("0".to_string()) - } + let bigint = element.as_bigint_ptr(); + let string = perry_runtime::bigint::js_bigint_to_string(bigint); + let value = crate::common::string_from_header_lossy(string) + .ok_or_else(|| format!("Could not read bigint at index {index}"))?; + ParamValue::String(value) } else if element.is_int32() { - ParamValue::Int(element.as_int32() as i64) + ParamValue::Int(i64::from(element.as_int32())) } else if element.is_bool() { ParamValue::Bool(element.as_bool()) } else if element.is_number() { - let n = element.to_number(); - // If the number is a whole number, send as Int for MySQL compatibility - // (MySQL prepared statements require integers for LIMIT, OFFSET, etc.) - if n.fract() == 0.0 && n >= i64::MIN as f64 && n <= i64::MAX as f64 { - ParamValue::Int(n as i64) + let number = element.to_number(); + if number.fract() == 0.0 && number >= i64::MIN as f64 && number <= i64::MAX as f64 { + ParamValue::Int(number as i64) } else { - ParamValue::Number(n) + ParamValue::Number(number) } } else { let mut byte_len = 0; @@ -452,373 +392,237 @@ pub(crate) unsafe fn extract_params_from_jsvalue( &mut byte_len, ); if !byte_ptr.is_null() { - let bytes = std::slice::from_raw_parts(byte_ptr, byte_len as usize); - ParamValue::Bytes(bytes.to_vec()) + ParamValue::Bytes(std::slice::from_raw_parts(byte_ptr, byte_len as usize).to_vec()) } else if perry_runtime::date::is_date_value(f64::from_bits(element_bits)) { let millis = perry_runtime::date::js_date_get_time(f64::from_bits(element_bits)); if !millis.is_finite() { - return Err(format!("Bind parameter at index {i} is an invalid Date")); + return Err(format!( + "Bind parameter at index {index} is an invalid Date" + )); } let date = chrono::DateTime::::from_timestamp_millis(millis as i64) .ok_or_else(|| { - format!("Bind parameter at index {i} is outside MySQL's Date range") + format!("Bind parameter at index {index} is outside MySQL's Date range") })? .naive_utc(); ParamValue::DateTime(date) } else { - return Err(format!("Unsupported bind parameter at index {i}")); + return Err(format!("Unsupported bind parameter at index {index}")); } }; - - result.push(param); + result.push(value); } - Ok(result) } -/// pool.getConnection() -> Promise -/// -/// Gets a connection from the pool. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_pool_get_connection(pool_handle: Handle) -> *mut Promise { +unsafe fn run_pool_query( + pool_handle: Handle, + query_f: f64, + params_f: f64, + force_prepared: bool, +) -> *mut Promise { let promise = js_promise_new_cross_thread(); - - crate::common::spawn_for_promise(promise as *mut u8, async move { - use crate::common::get_handle; - use tokio::time::timeout; - - if let Some(wrapper) = get_handle::(pool_handle) { - // Acquire a connection from the pool with timeout - match timeout( + let request = parse_query_request(query_f, params_f, force_prepared); + let pool = with_handle::(pool_handle, |wrapper| wrapper.pool.clone()); + let rows_as_array = request + .as_ref() + .map(|request| request.rows_as_array) + .unwrap_or(false); + + crate::common::spawn_for_promise_deferred_with_error( + promise as *mut u8, + async move { + let request = request?; + let pool = pool.ok_or_else(|| MysqlPromiseError::message("Invalid pool handle"))?; + // Pin a single physical connection for the complete operation. + let mut connection = tokio::time::timeout( Duration::from_secs(DEFAULT_ACQUIRE_TIMEOUT_SECS), - wrapper.pool.acquire(), + pool.acquire(), ) .await - { - Ok(Ok(conn)) => { - // Register the connection handle - let handle = register_handle(MysqlPoolConnectionHandle::new(conn)); - // NaN-box the handle with POINTER_TAG so it can be properly extracted later - // when conn.query() is called (codegen uses js_nanbox_get_pointer) - let nanboxed = perry_runtime::js_nanbox_pointer(handle as i64); - Ok(nanboxed.to_bits()) - } - Ok(Err(e)) => Err(format!("Failed to get connection: {}", e)), - Err(_) => Err(format!( - "Connection acquisition timed out after {} seconds", - DEFAULT_ACQUIRE_TIMEOUT_SECS - )), - } - } else { - Err("Invalid pool handle".to_string()) - } - }); - + .map_err(|_| MysqlPromiseError::message("Pool acquire timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Pool acquire failed", error))?; + execute_query_on_connection(&mut connection, &request).await + }, + move |outcome| outcome.to_jsvalue_with_rows_as_array(rows_as_array).bits(), + MysqlPromiseError::to_jsvalue_bits, + ); promise } -/// poolConnection.release() -/// -/// Returns a connection to the pool. -/// In sqlx, connections are automatically returned when dropped, -/// so we just need to drop the handle. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_pool_connection_release(conn_handle: Handle) { - // Enter the tokio runtime context before dropping the connection - // sqlx requires a runtime context when dropping pool connections - let _guard = crate::common::runtime().enter(); - - // Take and drop the connection handle - this releases the connection back to the pool - if let Some(_conn) = take_handle::(conn_handle) { - // Connection is automatically returned to pool when dropped - } -} - -/// poolConnection.query(sql, params?) -> Promise<[rows, fields]> -/// -/// Execute a query on the pool connection. See `js_mysql2_pool_query` for -/// rationale on accepting and binding `params` here. Issue #414. -#[no_mangle] -pub unsafe extern "C" fn js_mysql2_pool_connection_query( +unsafe fn run_pool_connection_query( conn_handle: Handle, - sql_ptr: *const u8, + query_f: f64, params_f: f64, + force_prepared: bool, ) -> *mut Promise { let promise = js_promise_new_cross_thread(); - let params = JSValue::from_bits(params_f.to_bits()); + let request = parse_query_request(query_f, params_f, force_prepared); + let connection = with_handle::(conn_handle, |wrapper| { + Arc::clone(&wrapper.connection) + }); + let rows_as_array = request + .as_ref() + .map(|request| request.rows_as_array) + .unwrap_or(false); - // Extract the SQL string - let sql = if sql_ptr.is_null() { - String::new() - } else { - let header = sql_ptr as *const perry_runtime::StringHeader; - let len = (*header).byte_len as usize; - let data_ptr = sql_ptr.add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - String::from_utf8_lossy(bytes).to_string() - }; + crate::common::spawn_for_promise_deferred_with_error( + promise as *mut u8, + async move { + let request = request?; + let connection = connection + .ok_or_else(|| MysqlPromiseError::message("Invalid pool connection handle"))?; + let mut slot = connection.lock().await; + let connection = slot + .as_mut() + .ok_or_else(|| MysqlPromiseError::message("Pool connection released"))?; + execute_query_on_connection(connection, &request).await + }, + move |outcome| outcome.to_jsvalue_with_rows_as_array(rows_as_array).bits(), + MysqlPromiseError::to_jsvalue_bits, + ); + promise +} - let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); +/// mysql.createPool(config) -> Pool. Like mysql2, construction is synchronous +/// and the first physical connection is opened lazily. +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_create_pool(config_f: f64) -> Handle { + let config = JSValue::from_bits(config_f.to_bits()); + let url = parse_mysql_config(config).to_url(); + let _runtime = crate::common::runtime().enter(); + MySqlPoolOptions::new() + .max_connections(10) + .acquire_timeout(Duration::from_secs(DEFAULT_ACQUIRE_TIMEOUT_SECS)) + .connect_lazy(&url) + .map(MysqlPoolHandle::new) + .map(register_handle) + .unwrap_or(0) +} - crate::common::spawn_for_promise_deferred( +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_pool_end(pool_handle: Handle) -> *mut Promise { + let promise = js_promise_new_cross_thread(); + let pool = take_handle::(pool_handle).map(|wrapper| wrapper.pool); + crate::common::spawn_for_promise_deferred_with_error( promise as *mut u8, async move { - use crate::common::get_handle_mut; - use tokio::time::timeout; - - let param_values = param_values?; - - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(ref mut conn) = wrapper.connection { - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } - - if is_select { - let query_future = query.fetch_all(&mut **conn); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(rows)) => { - let raw_result = RawQueryResult::from_mysql_rows(rows); - Ok(QueryOutcome::Rows(raw_result)) - } - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - let query_future = query.execute(&mut **conn); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(result)) => Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }), - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } - } else { - Err("Connection has been released".to_string()) - } - } else { - Err("Invalid connection handle".to_string()) - } + let pool = pool.ok_or_else(|| MysqlPromiseError::message("Invalid pool handle"))?; + let _ = tokio::time::timeout( + Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS), + pool.close(), + ) + .await; + Ok(JSValue::undefined().bits()) }, - |outcome: QueryOutcome| outcome.to_jsvalue().bits(), + |bits| bits, + MysqlPromiseError::to_jsvalue_bits, ); - promise } -/// poolConnection.execute(sql, params) -> Promise<[rows, fields]> -/// -/// Execute a prepared statement with parameters on the pool connection. #[no_mangle] -pub unsafe extern "C" fn js_mysql2_pool_connection_execute( - conn_handle: Handle, - sql_ptr: *const u8, +pub unsafe extern "C" fn js_mysql2_pool_query( + pool_handle: Handle, + query_f: f64, params_f: f64, ) -> *mut Promise { - let promise = js_promise_new_cross_thread(); - let params = JSValue::from_bits(params_f.to_bits()); - - // Extract the SQL string - let sql = if sql_ptr.is_null() { - String::new() - } else { - let header = sql_ptr as *const perry_runtime::StringHeader; - let len = (*header).byte_len as usize; - let data_ptr = sql_ptr.add(std::mem::size_of::()); - let bytes = std::slice::from_raw_parts(data_ptr, len); - String::from_utf8_lossy(bytes).to_string() - }; + run_pool_query(pool_handle, query_f, params_f, false) +} - // Extract parameters from the JSValue array - let param_values = extract_params_from_jsvalue(params); - let is_select = is_row_returning_query(&sql); +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_pool_execute( + pool_handle: Handle, + query_f: f64, + params_f: f64, +) -> *mut Promise { + run_pool_query(pool_handle, query_f, params_f, true) +} - crate::common::spawn_for_promise_deferred( +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_pool_get_connection(pool_handle: Handle) -> *mut Promise { + let promise = js_promise_new_cross_thread(); + let pool = with_handle::(pool_handle, |wrapper| wrapper.pool.clone()); + crate::common::spawn_for_promise_deferred_with_error( promise as *mut u8, async move { - use crate::common::get_handle_mut; - use tokio::time::timeout; - - let param_values = param_values?; - - if let Some(wrapper) = get_handle_mut::(conn_handle) { - if let Some(ref mut conn) = wrapper.connection { - // Build the query with parameter bindings - let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.clone())); - - for param in ¶m_values { - query = match param { - ParamValue::Null => query.bind(Option::::None), - ParamValue::String(s) => query.bind(s.clone()), - ParamValue::Bytes(bytes) => query.bind(bytes.clone()), - ParamValue::DateTime(date) => query.bind(*date), - ParamValue::Number(n) => query.bind(*n), - ParamValue::Int(i) => query.bind(*i), - ParamValue::Bool(b) => query.bind(*b), - }; - } - - if is_select { - let query_future = query.fetch_all(&mut **conn); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(rows)) => { - let raw_result = RawQueryResult::from_mysql_rows(rows); - Ok(QueryOutcome::Rows(raw_result)) - } - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } else { - let query_future = query.execute(&mut **conn); - match timeout( - Duration::from_secs(DEFAULT_QUERY_TIMEOUT_SECS), - query_future, - ) - .await - { - Ok(Ok(result)) => Ok(QueryOutcome::Executed { - affected_rows: result.rows_affected(), - last_insert_id: result.last_insert_id(), - }), - Ok(Err(e)) => Err(format!("Query failed: {}", e)), - Err(_) => Err(format!( - "Query timed out after {} seconds", - DEFAULT_QUERY_TIMEOUT_SECS - )), - } - } - } else { - Err("Connection has been released".to_string()) - } - } else { - Err("Invalid connection handle".to_string()) - } + let pool = pool.ok_or_else(|| MysqlPromiseError::message("Invalid pool handle"))?; + let connection = tokio::time::timeout( + Duration::from_secs(DEFAULT_ACQUIRE_TIMEOUT_SECS), + pool.acquire(), + ) + .await + .map_err(|_| MysqlPromiseError::message("Pool acquire timed out"))? + .map_err(|error| MysqlPromiseError::from_sqlx("Pool acquire failed", error))?; + Ok(connection) }, - |outcome: QueryOutcome| outcome.to_jsvalue().bits(), + |connection| { + let handle = register_handle(MysqlPoolConnectionHandle::new(connection)); + perry_runtime::js_nanbox_pointer(handle).to_bits() + }, + MysqlPromiseError::to_jsvalue_bits, ); - promise } +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_pool_connection_release(conn_handle: Handle) { + if let Some(wrapper) = take_handle::(conn_handle) { + crate::common::spawn(async move { + wrapper.connection.lock().await.take(); + }); + } +} + +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_pool_connection_query( + conn_handle: Handle, + query_f: f64, + params_f: f64, +) -> *mut Promise { + run_pool_connection_query(conn_handle, query_f, params_f, false) +} + +#[no_mangle] +pub unsafe extern "C" fn js_mysql2_pool_connection_execute( + conn_handle: Handle, + query_f: f64, + params_f: f64, +) -> *mut Promise { + run_pool_connection_query(conn_handle, query_f, params_f, true) +} + #[cfg(test)] mod tests { use super::*; - /// Issue #414: every shape the codegen dispatch table can pass to a - /// mysql2 query/execute params slot must be safely consumable. - /// - /// The dispatcher emits `args: &[NA_STR, NA_F64]` for both `db.query` - /// and `db.execute`, which in user code can be: - /// - missing (`db.query(sql)`) — codegen passes JS `undefined` - /// - undefined (`db.query(sql, undefined)`) — codegen passes TAG_UNDEFINED - /// - an array (`db.query(sql, [42])`) — codegen passes its NaN-boxed value - /// - a raw pointer (defensive compatibility with the old NA_PTR ABI) - /// - /// Pre-fix, only the no-params + execute-only paths were exercised; the - /// query path silently dropped a non-zero params arg because the FFI - /// signature only declared 2 args. Now query and execute share the - /// extract-and-bind path; these tests pin the four shapes. - #[test] - fn extract_params_returns_empty_for_codegen_no_args_pad() { - // Keep accepting the literal `0` used by the old NA_PTR ABI. - let v = unsafe { extract_params_from_jsvalue(JSValue::from_bits(0)) }.unwrap(); - assert!(v.is_empty(), "raw 0 must yield no params"); - } - #[test] - fn extract_params_returns_empty_for_undefined_and_null() { - let undef = - unsafe { extract_params_from_jsvalue(JSValue::from_bits(0x7FFC_0000_0000_0001)) } - .unwrap(); - assert!(undef.is_empty(), "TAG_UNDEFINED must yield no params"); - let null = - unsafe { extract_params_from_jsvalue(JSValue::from_bits(0x7FFC_0000_0000_0002)) } - .unwrap(); - assert!(null.is_empty(), "TAG_NULL must yield no params"); - } + fn text_protocol_is_used_only_for_query_without_values() { + let query = QueryRequest { + sql: "BEGIN".to_string(), + params: Vec::new(), + rows_as_array: false, + force_prepared: false, + }; + assert!(!query.uses_prepared_statement()); - #[test] - fn extract_params_handles_int_array_via_raw_pointer() { - unsafe { - let arr = perry_runtime::js_array_alloc(2); - let arr = perry_runtime::js_array_push_f64( - arr, - f64::from_bits(0x7FFE_0000_0000_002A), // INT32 42 - ); - let _arr = perry_runtime::js_array_push_f64( - arr, - f64::from_bits(0x7FFE_0000_0000_0001), // INT32 1 - ); + let execute = QueryRequest { + force_prepared: true, + ..query.clone() + }; + assert!(execute.uses_prepared_statement()); - // Codegen unboxes the NaN-boxed pointer to a raw i64. Mimic that - // by passing the raw lower-48-bits pointer (no tag). - let raw_ptr = arr as u64; - let v = extract_params_from_jsvalue(JSValue::from_bits(raw_ptr)).unwrap(); - assert_eq!(v.len(), 2, "should extract two int params"); - match &v[0] { - ParamValue::Int(n) => assert_eq!(*n, 42), - other => panic!("expected Int(42), got {:?}", other), - } - match &v[1] { - ParamValue::Int(n) => assert_eq!(*n, 1), - other => panic!("expected Int(1), got {:?}", other), - } - } + let parameterized = QueryRequest { + params: vec![ParamValue::Int(1)], + ..query + }; + assert!(parameterized.uses_prepared_statement()); } #[test] - fn extract_params_handles_int_array_via_nanboxed_pointer() { - unsafe { - let arr = perry_runtime::js_array_alloc(1); - let _arr = perry_runtime::js_array_push_f64( - arr, - f64::from_bits(0x7FFE_0000_0000_007B), // INT32 123 - ); - - // Defensive path: caller forgets to unbox before passing. - let nan_boxed = (arr as u64) | 0x7FFD_0000_0000_0000; - let v = extract_params_from_jsvalue(JSValue::from_bits(nan_boxed)).unwrap(); - assert_eq!(v.len(), 1); - match &v[0] { - ParamValue::Int(n) => assert_eq!(*n, 123), - other => panic!("expected Int(123), got {:?}", other), - } - } + fn mysql_error_code_names_match_mysql2() { + assert_eq!(mysql2_error_code(1062), Some("ER_DUP_ENTRY")); + assert_eq!(mysql2_error_code(1064), Some("ER_PARSE_ERROR")); + assert_eq!(mysql2_error_code(65_000), None); } } diff --git a/crates/perry-stdlib/src/mysql2/result.rs b/crates/perry-stdlib/src/mysql2/result.rs index 5f57affb17..897bdb12d1 100644 --- a/crates/perry-stdlib/src/mysql2/result.rs +++ b/crates/perry-stdlib/src/mysql2/result.rs @@ -105,14 +105,25 @@ impl RawQueryResult { /// Convert to JSValue (call this on main thread only!) pub fn to_jsvalue(&self) -> JSValue { + self.to_jsvalue_with_rows_as_array(false) + } + + /// Convert to mysql2's `[rows, fields]` tuple, optionally representing + /// every row as a positional array (`rowsAsArray: true`). + pub fn to_jsvalue_with_rows_as_array(&self, rows_as_array: bool) -> JSValue { // Create the result tuple [rows, fields] let mut result_array = js_array_alloc(2); // Create rows array let mut rows_array = js_array_alloc(self.rows.len() as u32); for row in &self.rows { - let row_obj = raw_row_to_js_object(row, &self.columns); - rows_array = js_array_push(rows_array, JSValue::object_ptr(row_obj as *mut u8)); + let row_value = if rows_as_array { + JSValue::array_ptr(raw_row_to_js_array(row)) + } else { + let row_obj = raw_row_to_js_object(row, &self.columns); + JSValue::object_ptr(row_obj as *mut u8) + }; + rows_array = js_array_push(rows_array, row_value); } let rows_jsval = JSValue::array_ptr(rows_array); result_array = js_array_push(result_array, rows_jsval); @@ -130,6 +141,16 @@ impl RawQueryResult { } } +/// Convert a raw row to the positional representation used by mysql2 when +/// `rowsAsArray` is enabled. +fn raw_row_to_js_array(row: &RawRowData) -> *mut perry_runtime::ArrayHeader { + let mut array = js_array_alloc(row.values.len() as u32); + for (_, value) in &row.values { + array = js_array_push(array, raw_value_to_jsvalue(value)); + } + array +} + /// Extract a raw value from a MySQL row (safe to call on any thread) fn extract_raw_value(row: &MySqlRow, index: usize, type_name: &str) -> RawValue { match type_name { @@ -346,8 +367,12 @@ pub fn rows_to_result_tuple(rows: Vec, columns: &[MySqlColumn]) -> JSV impl QueryOutcome { pub fn to_jsvalue(&self) -> JSValue { + self.to_jsvalue_with_rows_as_array(false) + } + + pub fn to_jsvalue_with_rows_as_array(&self, rows_as_array: bool) -> JSValue { match self { - QueryOutcome::Rows(raw) => raw.to_jsvalue(), + QueryOutcome::Rows(raw) => raw.to_jsvalue_with_rows_as_array(rows_as_array), QueryOutcome::Executed { affected_rows, last_insert_id, diff --git a/scripts/check_cross_thread_promise_provenance.py b/scripts/check_cross_thread_promise_provenance.py new file mode 100755 index 0000000000..c8e0d98c47 --- /dev/null +++ b/scripts/check_cross_thread_promise_provenance.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""#9552 — arena promises must not be handed to native settlement paths. + +A promise whose address leaves the runtime as a bare integer (a tokio future, +`std::thread::spawn`, a pending-result queue) is invisible to every root +scanner until its completion is queued back. Two constructors exist: + + * `js_promise_new_cross_thread()` — malloc-resident (non-moving) AND pinned + by the constructor until the promise settles (#9552). Safe to hand off. + * `js_promise_new()` / `js_promise_new_with_parent(..)` — nursery-resident. + A copying minor relocates it behind the worker's back, and a full + collection frees it. NEVER safe to hand off. + +This gate finds functions that mint a promise with an ARENA constructor and +pass it (directly, via `as usize`, or via a `let p = promise as usize` alias) +into a native settlement sink, or capture it in a spawned closure/future. +Sinks are matched by name, so a new hand-off API must be added to `SINKS`. + +Exit 1 on any hit. `--self-test` proves the detector can still fail: it plants +the bad shape (and its alias/spawn variants) and asserts each is reported, and +plants the good shape and asserts it is not. +""" +from __future__ import annotations + +import argparse +import pathlib +import re +import sys + +ROOT = pathlib.Path(__file__).resolve().parent.parent +SCAN_DIRS = ["crates/perry-runtime/src", "crates/perry-stdlib/src"] + [ + str(p.relative_to(ROOT)) for p in sorted(ROOT.glob("crates/perry-ext-*/src")) +] + +PROMISE_BINDING = re.compile( + r"\blet\s+(?:mut\s+)?(?P\w+)\s*(?::\s*[^=]+)?=\s*(?:[\w:]+::)?" + r"(?Pjs_promise_new(?:_with_parent|_cross_thread|_for_native_resolution)?)\s*\(" +) +ARENA_CTORS = {"js_promise_new", "js_promise_new_with_parent"} +ALIAS = re.compile(r"\blet\s+(?:mut\s+)?(?P\w+)\s*(?::\s*usize)?\s*=\s*(?P\w+)\s+as\s+usize\s*;") +SINKS = [ + "queue_promise_resolution", + "queue_deferred_resolution", + "queue_promise_string_result", + "queue_thread_result", + "spawn_for_promise", + "spawn_for_promise_deferred", + "perry_ffi_spawn_blocking", + "perry_ffi_spawn_async", + "spawn_blocking", + "spawn", +] +SINK_CALL = re.compile(r"\b(?:[\w:]+::)?(?P" + "|".join(map(re.escape, SINKS)) + r")\s*\(") +FN_HEADER = re.compile(r"\bfn\s+(?P\w+)\s*(?:<[^>]*>)?\s*\(") + + +def strip_comments(src: str) -> str: + """Blank out `//` comments (keeping line structure) so doc prose cannot + trip the matchers.""" + out = [] + for line in src.split("\n"): + cut = line.find("//") + if cut != -1 and line.count('"', 0, cut) % 2 == 0: + line = line[:cut] + out.append(line) + return "\n".join(out) + + +def match_brace(src: str, open_idx: int) -> int: + """Index just past the `}` matching the `{` at `open_idx` (or the `)` for + `(`), ignoring string literals.""" + opener = src[open_idx] + closer = {"{": "}", "(": ")"}[opener] + depth = 0 + i = open_idx + in_str = False + while i < len(src): + c = src[i] + if in_str: + if c == "\\": + i += 2 + continue + if c == '"': + in_str = False + elif c == '"': + in_str = True + elif c == opener: + depth += 1 + elif c == closer: + depth -= 1 + if depth == 0: + return i + 1 + i += 1 + return len(src) + + +def function_bodies(src: str): + """Yield (fn_name, body_start, body_text) for every fn in `src`.""" + for m in FN_HEADER.finditer(src): + sig_end = match_brace(src, m.end() - 1) + brace = src.find("{", sig_end) + semi = src.find(";", sig_end) + if brace == -1 or (semi != -1 and semi < brace): + continue # trait method without a body + end = match_brace(src, brace) + yield m.group("fn"), brace, src[brace:end] + + +def latest_before(entries, name, pos): + """The last (pos, value) entry for `name` that precedes `pos`, or None. + Bindings shadow: an early-return arena promise named `promise` must not + taint a later `let promise = js_promise_new_cross_thread()`.""" + best = None + for entry_pos, value in entries.get(name, ()): + if entry_pos < pos and (best is None or entry_pos > best[0]): + best = (entry_pos, value) + return best + + +def analyze_source(src: str, path: str = ""): + """Return a list of (line, fn, message) findings for one Rust source.""" + src = strip_comments(src) + findings = [] + for fn_name, body_start, body in function_bodies(src): + bindings = {} + for m in PROMISE_BINDING.finditer(body): + bindings.setdefault(m.group("name"), []).append((m.start(), m.group("ctor"))) + if not any(ctor in ARENA_CTORS for entries in bindings.values() for _, ctor in entries): + continue + aliases = {} + for m in ALIAS.finditer(body): + aliases.setdefault(m.group("alias"), []).append((m.start(), m.group("src"))) + names = set(bindings) | set(aliases) + ident = re.compile(r"\b(" + "|".join(map(re.escape, sorted(names))) + r")\b") + for call in SINK_CALL.finditer(body): + args_end = match_brace(body, call.end() - 1) + args = body[call.end() - 1 : args_end] + for hit in ident.finditer(args): + name, pos = hit.group(1), call.start() + alias = latest_before(aliases, name, pos) + if alias is not None: + pos, name = alias + binding = latest_before(bindings, name, pos) + if binding is None or binding[1] not in ARENA_CTORS: + continue + line = src.count("\n", 0, body_start + call.start()) + 1 + findings.append( + ( + line, + fn_name, + f"arena promise `{hit.group(1)}` (from {binding[1]}) reaches native sink " + f"`{call.group('sink')}`; mint it with js_promise_new_cross_thread() instead", + ) + ) + break + return findings + + +def scan_tree(): + findings = [] + for rel in SCAN_DIRS: + for path in sorted((ROOT / rel).rglob("*.rs")): + for line, fn_name, msg in analyze_source(path.read_text(), str(path)): + findings.append(f"{path.relative_to(ROOT)}:{line}: in `{fn_name}`: {msg}") + return findings + + +BAD_DIRECT = """ +pub unsafe extern "C" fn bad_direct() -> *mut Promise { + let promise = perry_runtime::js_promise_new(); + queue_promise_resolution(promise as usize, true, 0); + promise +} +""" +BAD_ALIAS_SPAWN = """ +pub unsafe extern "C" fn bad_alias() -> *mut Promise { + let promise = js_promise_new(); + let promise_ptr = promise as usize; + spawn(async move { + queue_deferred_resolution(promise_ptr, true, || 0); + }); + promise +} +""" +BAD_WITH_PARENT = """ +fn bad_parent(parent: *mut Promise) { + let p = crate::promise::js_promise_new_with_parent(parent); + let raw = p as usize; + std::thread::spawn(move || queue_promise_string_result(0, raw, String::new())); +} +""" +GOOD_CROSS_THREAD = """ +pub unsafe extern "C" fn good() -> *mut Promise { + let promise = perry_runtime::js_promise_new_cross_thread(); + let promise_ptr = promise as usize; + spawn(async move { queue_promise_resolution(promise_ptr, true, 0); }); + promise +} +""" +GOOD_SHADOWED = """ +unsafe fn good_shadowed(closure: *const u8) -> *mut Promise { + if closure.is_null() { + let promise = crate::promise::js_promise_new(); + crate::promise::js_promise_resolve(promise, 0.0); + return promise; + } + let promise = crate::promise::js_promise_new_cross_thread(); + let promise_usize = promise as usize; + std::thread::spawn(move || queue_thread_result(0, promise_usize, 0)); + promise +} +""" +GOOD_UNRELATED = """ +fn good_unrelated(writable_id: usize) -> *mut Promise { + let promise = js_promise_new(); + // a different usize reaches the sink; the promise stays on the main thread + let job = writable_id as usize; + spawn(async move { finish(job) }); + promise +} +""" + + +def self_test() -> int: + failures = [] + for label, snippet, expect in [ + ("direct", BAD_DIRECT, True), + ("alias+spawn", BAD_ALIAS_SPAWN, True), + ("with_parent+thread::spawn", BAD_WITH_PARENT, True), + ("cross_thread ctor", GOOD_CROSS_THREAD, False), + ("unrelated usize", GOOD_UNRELATED, False), + ("shadowed early-return arena promise", GOOD_SHADOWED, False), + ]: + got = bool(analyze_source(snippet)) + if got != expect: + failures.append(f"{label}: expected {'a hit' if expect else 'no hit'}, got {analyze_source(snippet)}") + if failures: + print("check_cross_thread_promise_provenance --self-test FAILED:") + for f in failures: + print(" " + f) + return 1 + print("check_cross_thread_promise_provenance --self-test ok (3 planted shapes caught, 3 clean shapes pass)") + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--self-test", action="store_true") + args = ap.parse_args() + if args.self_test: + return self_test() + findings = scan_tree() + if findings: + print("#9552 cross-thread promise provenance: arena promises handed to native settlement paths:") + for f in findings: + print(" " + f) + print(f"{len(findings)} finding(s). Mint with js_promise_new_cross_thread() (pinned until settled).") + return 1 + print("check_cross_thread_promise_provenance: no arena promise reaches a native settlement sink") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test-files/test_gap_9552_cross_thread_promise_survives_gc.ts b/test-files/test_gap_9552_cross_thread_promise_survives_gc.ts new file mode 100644 index 0000000000..df6a4b10c5 --- /dev/null +++ b/test-files/test_gap_9552_cross_thread_promise_survives_gc.ts @@ -0,0 +1,44 @@ +// #9552: a promise minted for a cross-thread settlement — every stdlib +// `fetch` response, among ~110 stdlib call sites — is referenced only by the +// worker's raw address while the request is in flight: the awaiting +// continuation hangs OFF the promise (`P.on_fulfilled`), nothing on the JS +// side points AT it. A full collection landing in that window freed it, and +// the completion then resolved whatever the allocator had reused the slot for +// (the report: a RegExp header read as a promise's `next`, SIGSEGV in the +// microtask pump). +// +// The constructor now pins the promise until it settles. This runs a request +// against a local server that answers late, forces collections while it is in +// flight, and expects the response to arrive. Node only exposes `gc` under +// --expose-gc, so the collection is conditional and the expected output is +// identical on both. +import http from "node:http"; + +declare const gc: undefined | (() => void); + +const server = http.createServer((_req, res) => { + setTimeout(() => { + res.end("ok"); + }, 60); +}); + +async function get(url: string): Promise { + const response = await fetch(url); + return await response.text(); +} + +server.listen(0, "127.0.0.1", async () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + const inflight = get(`http://127.0.0.1:${port}/`); + + let junk: Array<{ k: number; s: string }> = []; + for (let round = 0; round < 8; round++) { + junk = Array.from({ length: 4000 }, (_, k) => ({ k, s: "x".repeat(40) })); + if (typeof gc === "function") gc(); + await new Promise((resolve) => setTimeout(resolve, 5)); + } + + console.log(await inflight, junk.length); + server.close(); +}); diff --git a/test-files/test_issue_9330_9517_mysql2_bundled.ts b/test-files/test_issue_9330_9517_mysql2_bundled.ts new file mode 100644 index 0000000000..b2f012f9b5 --- /dev/null +++ b/test-files/test_issue_9330_9517_mysql2_bundled.ts @@ -0,0 +1,101 @@ +// parity-skip: requires a live MySQL fixture and the bundled mysql2 fallback +// Regression coverage for #9330 and #9517. Run with DB_HOST, DB_PORT, +// DB_USER, DB_PASSWORD, and DB_NAME set, and compile without Cargo on PATH so +// Perry selects its bundled mysql2 implementation. +// platforms: skip + +import mysql from 'mysql2/promise'; + +function assert(condition: unknown, message: string): asserts condition { + if (!condition) throw new Error(message); +} + +async function main(): Promise { + const pool: any = mysql.createPool({ + host: process.env.DB_HOST ?? '127.0.0.1', + port: Number(process.env.DB_PORT ?? 3306), + user: process.env.DB_USER, + password: process.env.DB_PASSWORD, + database: process.env.DB_NAME, + }); + + assert('getConnection' in pool, 'getConnection must be visible to `in`'); + assert(Reflect.has(pool, 'getConnection'), 'getConnection must be reflectable'); + assert(typeof pool.getConnection === 'function', 'getConnection must be callable'); + + // A dynamically typed receiver must still route to mysql2 instead of being + // interpreted as a raw object pointer. + const dynamic: any = pool; + const [dynamicRows] = await dynamic.query('SELECT 9517 AS issue'); + assert(dynamicRows[0].issue === 9517, 'dynamic query returned the wrong row'); + + // The options-object form is used by Drizzle. `values` is accepted from the + // object and rowsAsArray changes the row representation. + const [arrayRows] = await pool.query({ + sql: 'SELECT ? AS issue', + values: [9330], + rowsAsArray: true, + }); + assert(Array.isArray(arrayRows[0]), 'rowsAsArray did not return positional rows'); + assert(arrayRows[0][0] === 9330, 'options.values was not bound'); + + // mysql2 query() without values uses the text protocol. MySQL rejects BEGIN + // when it is sent through COM_STMT_PREPARE. + await pool.query('BEGIN'); + await pool.query('ROLLBACK'); + + let syntaxError: any; + try { + await pool.query('SELEC broken syntax'); + } catch (error: any) { + syntaxError = error; + } + assert(syntaxError instanceof Error, 'query rejection must be an Error'); + assert(typeof syntaxError.message === 'string', 'query Error needs .message'); + assert(syntaxError.code === 'ER_PARSE_ERROR', 'query Error needs mysql2 .code'); + assert(syntaxError.errno === 1064, 'query Error needs mysql2 .errno'); + + const getConnection = pool.getConnection; + if (process.env.DB_USE_EXISTING_NOTIFICATIONS === '1') { + // Useful on constrained test hosts where the fixture table already exists + // but creating another InnoDB tablespace is not possible. + const [beforeRows] = await pool.query( + 'SELECT id, body FROM notifications ORDER BY id LIMIT 1', + ); + assert(beforeRows.length === 1, 'notifications fixture needs one row'); + const before = beforeRows[0]; + const connection: any = await getConnection(); + try { + await connection.beginTransaction(); + await connection.execute('UPDATE notifications SET body = ? WHERE id = ?', [ + 'tx-9330', + before.id, + ]); + await connection.rollback(); + } finally { + connection.release(); + } + const [afterRows] = await pool.query('SELECT body FROM notifications WHERE id = ?', [ + before.id, + ]); + assert(afterRows[0].body === before.body, 'rollback escaped its checked-out connection'); + } else { + await pool.query('DROP TABLE IF EXISTS perry_issue_9330'); + await pool.query('CREATE TABLE perry_issue_9330 (value INT NOT NULL)'); + const connection: any = await getConnection(); + try { + await connection.beginTransaction(); + await connection.execute('INSERT INTO perry_issue_9330 (value) VALUES (?)', [1]); + await connection.rollback(); + } finally { + connection.release(); + } + const [countRows] = await pool.query('SELECT COUNT(*) AS count FROM perry_issue_9330'); + assert(countRows[0].count === 0, 'rollback escaped its checked-out connection'); + } + + await pool.end(); + console.log('issues 9330/9517 mysql2 bundled: OK'); +} + +main(); diff --git a/test-files/test_issue_9330_drizzle_mysql2_transaction.ts b/test-files/test_issue_9330_drizzle_mysql2_transaction.ts new file mode 100644 index 0000000000..7f88c21ce0 --- /dev/null +++ b/test-files/test_issue_9330_drizzle_mysql2_transaction.ts @@ -0,0 +1,54 @@ +// parity-skip: requires drizzle-orm, mysql2, and a live MySQL fixture +// End-to-end regression for #9330. Compile with Cargo absent from PATH to +// exercise the bundled sqlx-backed mysql2 implementation. +// platforms: skip + +import mysql from 'mysql2/promise'; +import { drizzle } from 'drizzle-orm/mysql2'; +import { sql } from 'drizzle-orm'; + +function assert(condition: unknown, message: string): asserts condition { + if (!condition) throw new Error(message); +} + +function connectionId(result: any): number { + const rows = Array.isArray(result) ? result[0] : result; + return Number(rows[0].id); +} + +const pool: any = mysql.createPool({ + host: process.env.DB_HOST ?? '127.0.0.1', + port: Number(process.env.DB_PORT ?? 3306), + user: process.env.DB_USER, + password: process.env.DB_PASSWORD, + database: process.env.DB_NAME, + // These mysql2-only keys select Perry's native implementation. + connectionLimit: 4, + waitForConnections: true, +}); +const db: any = drizzle(pool, { mode: 'default' }); + +assert('getConnection' in pool, 'Drizzle must recognize the mysql2 pool'); +const ids: number[] = []; +await db.transaction(async (tx: any) => { + ids.push(connectionId(await tx.execute(sql`SELECT CONNECTION_ID() AS id`))); + ids.push(connectionId(await tx.execute(sql`SELECT CONNECTION_ID() AS id`))); + ids.push(connectionId(await tx.execute(sql`SELECT CONNECTION_ID() AS id`))); + // Acquire a real row lock without changing fixture data. + await tx.execute(sql`UPDATE notifications SET body = body ORDER BY id LIMIT 1`); +}); + +assert(ids[0] > 0, 'transaction did not return a connection id'); +assert(ids.every((id) => id === ids[0]), `transaction scattered across: ${ids.join(',')}`); +if (process.env.DB_CHECK_INNODB_TRX === '1') { + // Requires MySQL's PROCESS privilege; connection-id stability above is the + // portable atomicity assertion, while this additionally catches lock leaks. + const [openTransactions] = await pool.query( + 'SELECT trx_id FROM information_schema.innodb_trx WHERE trx_mysql_thread_id = ?', + [ids[0]], + ); + assert(openTransactions.length === 0, 'transaction connection remained open after commit'); +} + +await pool.end(); +console.log('issue 9330 drizzle mysql2 transaction: OK');