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-stdlib/src/common/async_bridge.rs b/crates/perry-stdlib/src/common/async_bridge.rs index 86e1e42157..2cd1a801b9 100644 --- a/crates/perry-stdlib/src/common/async_bridge.rs +++ b/crates/perry-stdlib/src/common/async_bridge.rs @@ -1053,6 +1053,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/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');