diff --git a/crates/weaveffi/tests/runtime.rs b/crates/weaveffi/tests/runtime.rs index f4beaf4..b6e8002 100644 --- a/crates/weaveffi/tests/runtime.rs +++ b/crates/weaveffi/tests/runtime.rs @@ -451,6 +451,28 @@ fn decode_ret(ptr: *const u8, len: usize) -> T { value } +/// Build the `context` for one async launch: a heap-boxed clone of the +/// test's channel sender that the completion callback adopts with +/// [`adopt_ctx`]. +/// +/// The callback fires on a detached producer thread, and the test's `recv` +/// unblocks as soon as the message is queued, before that thread has finished +/// `send`. Handing the callback a *borrowed* sender and freeing it after +/// `recv` would therefore free the channel out from under an in-flight +/// `send`. Giving every launch its own owned sender, released by the callback +/// itself, means the channel is torn down by whichever side lets go last, the +/// way the ABI's "consumer owns `context`" rule is meant to be used. +fn ctx_for(tx: &std::sync::mpsc::Sender) -> *mut std::os::raw::c_void { + Box::into_raw(Box::new(tx.clone())).cast() +} + +/// Take ownership of the sender [`ctx_for`] boxed for this callback. Must be +/// called exactly once per launch, which the "callback fires exactly once" +/// contract guarantees. +unsafe fn adopt_ctx(ctx: *mut std::os::raw::c_void) -> Box> { + unsafe { Box::from_raw(ctx.cast()) } +} + #[test] fn scalar_call_sets_ok() { let mut err = ok_err(); @@ -932,7 +954,7 @@ fn callback_interface_from_async_method() { type Msg = (i32, String, i64); extern "C" fn cb(ctx: *mut c_void, err: *mut weaveffi_error, result: i64) { - let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(ctx) }; let (code, msg) = if err.is_null() { (0, String::new()) } else { @@ -950,16 +972,15 @@ fn callback_interface_from_async_method() { bus::weaveffi_bus_Bus_subscribe(b, new_ctx(4, &freed), &VTABLE, &mut err); let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); let text = string_to_c_ptr("async"); - bus::weaveffi_bus_Bus_publish_later_async(b, text, 2, cb, tx_ptr as *mut c_void); + bus::weaveffi_bus_Bus_publish_later_async(b, text, 2, cb, ctx_for(&tx)); let (code, _, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!(code, 0); assert_eq!(result, 2); // A foreign failure inside the future is delivered through the callback. - bus::weaveffi_bus_Bus_publish_later_async(b, text, 4, cb, tx_ptr as *mut c_void); + bus::weaveffi_bus_Bus_publish_later_async(b, text, 4, cb, ctx_for(&tx)); let (code, msg, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!(code, abi::FOREIGN_ERROR_CODE); assert_eq!(msg, "subscriber rejected async"); @@ -967,14 +988,13 @@ fn callback_interface_from_async_method() { // The receiver was retained across the spawn: releasing the consumer's // reference while a call is in flight is safe. - bus::weaveffi_bus_Bus_publish_later_async(b, text, 1, cb, tx_ptr as *mut c_void); + bus::weaveffi_bus_Bus_publish_later_async(b, text, 1, cb, ctx_for(&tx)); bus::weaveffi_bus_Bus_destroy(b); let (code, _, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!(code, 0); assert_eq!(result, 3); free_string(text); - unsafe { drop(Box::from_raw(tx_ptr)) }; } #[test] @@ -992,7 +1012,7 @@ fn async_struct_result_completes_via_callback() { result_ptr: *const u8, result_len: usize, ) { - let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(ctx) }; let had_err = !err.is_null() && unsafe { (*err).code } != 0; if had_err { abi::error_free(err); @@ -1009,12 +1029,10 @@ fn async_struct_result_completes_via_callback() { } let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); let name = string_to_c_ptr("alpha"); - tasks::weaveffi_tasks_run_task_async(name, cb, tx_ptr as *mut c_void); + tasks::weaveffi_tasks_run_task_async(name, cb, ctx_for(&tx)); let (had_err, id, value) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); free_string(name); - unsafe { drop(Box::from_raw(tx_ptr)) }; assert!(!had_err); assert_eq!(id, 7); @@ -1029,7 +1047,7 @@ fn async_result_ok_and_err_paths() { type Msg = (bool, i32); extern "C" fn cb(ctx: *mut c_void, err: *mut weaveffi_error, result: i32) { - let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(ctx) }; let had_err = !err.is_null() && unsafe { (*err).code } != 0; if had_err { // The reported error is heap-boxed and owned by the consumer. @@ -1039,19 +1057,16 @@ fn async_result_ok_and_err_paths() { } let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); - tasks::weaveffi_tasks_checked_add_async(2, 3, cb, tx_ptr as *mut c_void); + tasks::weaveffi_tasks_checked_add_async(2, 3, cb, ctx_for(&tx)); let (had_err, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert!(!had_err); assert_eq!(result, 5); - tasks::weaveffi_tasks_checked_add_async(i32::MAX, 1, cb, tx_ptr as *mut c_void); + tasks::weaveffi_tasks_checked_add_async(i32::MAX, 1, cb, ctx_for(&tx)); let (had_err, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert!(had_err); assert_eq!(result, 0); - - unsafe { drop(Box::from_raw(tx_ptr)) }; } #[test] @@ -1088,7 +1103,7 @@ fn deferred_foreign_error_fires_the_async_callback_once() { type Msg = (i32, String, i32); extern "C" fn cb(ctx: *mut c_void, err: *mut weaveffi_error, result: i32) { - let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(ctx) }; let (code, message) = if err.is_null() { (0, String::new()) } else { @@ -1101,13 +1116,12 @@ fn deferred_foreign_error_fires_the_async_callback_once() { } let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); - deferred::weaveffi_deferred_later_then_fail_async(false, cb, tx_ptr as *mut c_void); + deferred::weaveffi_deferred_later_then_fail_async(false, cb, ctx_for(&tx)); let (code, _, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!((code, result), (0, 88)); - deferred::weaveffi_deferred_later_then_fail_async(true, cb, tx_ptr as *mut c_void); + deferred::weaveffi_deferred_later_then_fail_async(true, cb, ctx_for(&tx)); let (code, message, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!(code, abi::FOREIGN_ERROR_CODE); assert_eq!(message, "consumer said no, later"); @@ -1116,8 +1130,6 @@ fn deferred_foreign_error_fires_the_async_callback_once() { rx.recv_timeout(Duration::from_millis(200)).is_err(), "the completion callback fires exactly once" ); - - unsafe { drop(Box::from_raw(tx_ptr)) }; } /// Exercises nested-module codegen: the inner module's symbols must carry the @@ -1644,7 +1656,7 @@ fn async_methods_retain_the_receiver() { use std::time::Duration; extern "C" fn on_value(ctx: *mut c_void, err: *mut weaveffi_error, result: i64) { - let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(ctx) }; assert!(err.is_null()); tx.send(result).unwrap(); } @@ -1653,7 +1665,7 @@ fn async_methods_retain_the_receiver() { err: *mut weaveffi_error, result: *mut counters::Counter, ) { - let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(ctx) }; assert!(err.is_null()); let mut e = weaveffi_error::default(); let v = counters::weaveffi_counters_Counter_value(result, &mut e); @@ -1663,11 +1675,10 @@ fn async_methods_retain_the_receiver() { let mut err = ok_err(); let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); let c = counters::weaveffi_counters_Counter_new(8, &mut err); - counters::weaveffi_counters_Counter_value_later_async(c, on_value, tx_ptr as *mut c_void); - counters::weaveffi_counters_Counter_snapshot_later_async(c, on_obj, tx_ptr as *mut c_void); + counters::weaveffi_counters_Counter_value_later_async(c, on_value, ctx_for(&tx)); + counters::weaveffi_counters_Counter_snapshot_later_async(c, on_obj, ctx_for(&tx)); // Releasing the consumer's reference while calls are in flight is safe: // each launcher retained its own. counters::weaveffi_counters_Counter_destroy(c); @@ -1680,19 +1691,14 @@ fn async_methods_retain_the_receiver() { // A null receiver still completes (with a marshalling error). extern "C" fn on_null(ctx: *mut c_void, err: *mut weaveffi_error, result: i64) { - let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(ctx) }; assert!(!err.is_null()); assert_eq!(unsafe { (*err).code }, abi::MARSHAL_ERROR_CODE); abi::error_free(err); tx.send(result).unwrap(); } - counters::weaveffi_counters_Counter_value_later_async( - std::ptr::null(), - on_null, - tx_ptr as *mut c_void, - ); + counters::weaveffi_counters_Counter_value_later_async(std::ptr::null(), on_null, ctx_for(&tx)); assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 0); - unsafe { drop(Box::from_raw(tx_ptr)) }; } #[test] diff --git a/samples/async-demo/src/lib.rs b/samples/async-demo/src/lib.rs index 1e00ed1..f7de040 100644 --- a/samples/async-demo/src/lib.rs +++ b/samples/async-demo/src/lib.rs @@ -137,7 +137,7 @@ mod tests { result_ptr: *const u8, result_len: usize, ) { - let tx = unsafe { &*(context as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(context) }; let code = if err.is_null() { 0 } else { @@ -158,7 +158,7 @@ mod tests { results_ptr: *const u8, results_len: usize, ) { - let tx = unsafe { &*(context as *const mpsc::Sender) }; + let tx = unsafe { adopt_ctx::(context) }; let had_error = !err.is_null() && unsafe { (*err).code } != 0; // Decode the borrowed value buffer before returning; the launcher // frees it once the callback completes. @@ -172,35 +172,39 @@ mod tests { } extern "C" fn n_tasks_callback(context: *mut c_void, err: *mut weaveffi_error, result: i32) { - let tx = unsafe { &*(context as *const mpsc::Sender<(bool, i32)>) }; + let tx = unsafe { adopt_ctx::<(bool, i32)>(context) }; let had_error = !err.is_null() && unsafe { (*err).code } != 0; let _ = tx.send((had_error, result)); } - /// Intentionally leak a callback-context box. + /// Box a clone of the test's sender as the `context` for one async + /// launch, to be adopted by the completion callback with [`adopt_ctx`]. /// /// The `#[weaveffi::module]` async launchers invoke the completion callback - /// on a detached worker thread, so a worker may still be inside the - /// callback's `send` when the test's `recv` returns (the receiver unblocks - /// as soon as the message is queued, before `send` finishes). Reclaiming the - /// box here would free the `Sender` out from under that in-flight `send`, a - /// use-after-free. The test deliberately leaks the context instead, which - /// keeps the channel alive for the brief remaining life of any in-flight - /// callback; the OS reclaims the memory at process exit. - fn leak_ctx(ptr: *mut T) { - std::mem::forget(unsafe { Box::from_raw(ptr) }); + /// on a detached worker thread, and the test's `recv` unblocks as soon as + /// the message is queued, before that worker has finished `send`. Handing + /// the callback a borrowed sender and freeing it after `recv` would + /// therefore free the channel out from under an in-flight `send`. Giving + /// each launch its own owned sender, released by the callback itself, + /// means the channel is torn down by whichever side lets go last. + fn ctx_for(tx: &mpsc::Sender) -> *mut c_void { + Box::into_raw(Box::new(tx.clone())).cast() + } + + /// Take ownership of the sender [`ctx_for`] boxed for this callback. The + /// "callback fires exactly once" contract makes this a single adoption. + unsafe fn adopt_ctx(ctx: *mut c_void) -> Box> { + unsafe { Box::from_raw(ctx.cast()) } } #[test] fn run_task_calls_callback() { let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); let name = CString::new("test-task").unwrap(); - weaveffi_tasks_run_task_async(name.as_ptr(), task_callback, tx_ptr as *mut c_void); + weaveffi_tasks_run_task_async(name.as_ptr(), task_callback, ctx_for(&tx)); let (code, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); - leak_ctx(tx_ptr); assert_eq!(code, 0); let r = result.expect("success path passes a result buffer"); @@ -212,13 +216,11 @@ mod tests { #[test] fn run_task_empty_name_reports_invalid_name() { let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); let empty = CString::new("").unwrap(); - weaveffi_tasks_run_task_async(empty.as_ptr(), task_callback, tx_ptr as *mut c_void); + weaveffi_tasks_run_task_async(empty.as_ptr(), task_callback, ctx_for(&tx)); let (code, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); - leak_ctx(tx_ptr); assert_eq!(code, 1, "TaskError::InvalidName's declared code"); assert!(result.is_none(), "error path passes a null result buffer"); } @@ -226,15 +228,13 @@ mod tests { #[test] fn run_task_null_name_reports_marshal_error_through_the_callback() { let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); // The launcher has no out_err slot, so an argument that fails to lift // is reported through the completion callback with the reserved // marshalling code, exactly like the sync path would report it. - weaveffi_tasks_run_task_async(std::ptr::null(), task_callback, tx_ptr as *mut c_void); + weaveffi_tasks_run_task_async(std::ptr::null(), task_callback, ctx_for(&tx)); let (code, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); - leak_ctx(tx_ptr); assert_eq!(code, abi::MARSHAL_ERROR_CODE); assert!(result.is_none()); } @@ -242,7 +242,6 @@ mod tests { #[test] fn run_batch_processes_sequentially() { let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); // The list-of-strings parameter is buffered: encode `Vec` // and pass the (ptr, len) pair. The launcher copies the bytes before // returning, so the local buffer only needs to outlive the call. @@ -252,15 +251,9 @@ mod tests { "task-c".to_string(), ]); - weaveffi_tasks_run_batch_async( - names.as_ptr(), - names.len(), - batch_callback, - tx_ptr as *mut c_void, - ); + weaveffi_tasks_run_batch_async(names.as_ptr(), names.len(), batch_callback, ctx_for(&tx)); let (had_error, results) = rx.recv_timeout(Duration::from_secs(10)).unwrap(); - leak_ctx(tx_ptr); assert!(!had_error); assert_eq!(results.len(), 3); @@ -276,18 +269,11 @@ mod tests { #[test] fn run_batch_empty_names() { let (tx, rx) = mpsc::channel::(); - let tx_ptr = Box::into_raw(Box::new(tx)); let names = abi::encode_value(&Vec::::new()); - weaveffi_tasks_run_batch_async( - names.as_ptr(), - names.len(), - batch_callback, - tx_ptr as *mut c_void, - ); + weaveffi_tasks_run_batch_async(names.as_ptr(), names.len(), batch_callback, ctx_for(&tx)); let (had_error, results) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); - leak_ctx(tx_ptr); assert!(!had_error); assert!(results.is_empty()); } @@ -331,10 +317,8 @@ mod tests { #[test] fn run_n_tasks_invokes_callback_with_n() { let (tx, rx) = mpsc::channel::<(bool, i32)>(); - let tx_ptr = Box::into_raw(Box::new(tx)); - weaveffi_tasks_run_n_tasks_async(7, n_tasks_callback, tx_ptr as *mut c_void); + weaveffi_tasks_run_n_tasks_async(7, n_tasks_callback, ctx_for(&tx)); let (had_error, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); - leak_ctx(tx_ptr); assert!(!had_error); assert_eq!(result, 7); } @@ -343,15 +327,13 @@ mod tests { fn active_callbacks_returns_to_zero() { let mut err = weaveffi_error::default(); let (tx, rx) = mpsc::channel::<(bool, i32)>(); - let tx_ptr = Box::into_raw(Box::new(tx)); for i in 0..16 { - weaveffi_tasks_run_n_tasks_async(i, n_tasks_callback, tx_ptr as *mut c_void); + weaveffi_tasks_run_n_tasks_async(i, n_tasks_callback, ctx_for(&tx)); } for _ in 0..16 { rx.recv_timeout(Duration::from_secs(5)).unwrap(); } - leak_ctx(tx_ptr); for _ in 0..50 { if weaveffi_tasks_active_callbacks(&mut err) == 0 { diff --git a/samples/kvstore/src/lib.rs b/samples/kvstore/src/lib.rs index 7749a21..e346199 100644 --- a/samples/kvstore/src/lib.rs +++ b/samples/kvstore/src/lib.rs @@ -474,6 +474,19 @@ mod tests { weaveffi_error::default() } + /// Box a clone of the test's sender as the `context` for one async + /// launch. The completion callback adopts it with [`adopt_ctx`], so the + /// channel is released by whichever side lets go last; a borrowed sender + /// freed after `recv` would race the producer thread's in-flight `send`. + fn ctx_for(tx: &mpsc::Sender) -> *mut c_void { + Box::into_raw(Box::new(tx.clone())).cast() + } + + /// Take ownership of the sender [`ctx_for`] boxed for this callback. + unsafe fn adopt_ctx(ctx: *mut c_void) -> Box> { + unsafe { Box::from_raw(ctx.cast()) } + } + fn open() -> *mut Store { let mut err = new_err(); let path = CString::new("/tmp/kvstore-test").unwrap(); @@ -774,9 +787,8 @@ mod tests { ); let (tx, rx) = mpsc::channel::<(i32, i64)>(); - let tx_ptr = Box::into_raw(Box::new(tx)); extern "C" fn cb(context: *mut c_void, err: *mut weaveffi_error, result: i64) { - let tx = unsafe { &*(context as *const mpsc::Sender<(i32, i64)>) }; + let tx = unsafe { adopt_ctx::<(i32, i64)>(context) }; let code = if err.is_null() { 0 } else { @@ -785,10 +797,9 @@ mod tests { tx.send((code, result)).unwrap(); } let token = abi::cancel_token_create(); - weaveffi_kv_Store_compact_async(s, token, cb, tx_ptr as *mut c_void); + weaveffi_kv_Store_compact_async(s, token, cb, ctx_for(&tx)); let (code, reclaimed) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); - unsafe { drop(Box::from_raw(tx_ptr)) }; assert_eq!(code, 0); assert_eq!(reclaimed, 5); abi::cancel_token_destroy(token); @@ -801,9 +812,8 @@ mod tests { let s = open(); let (tx, rx) = mpsc::channel::<(i32, i64)>(); - let tx_ptr = Box::into_raw(Box::new(tx)); extern "C" fn cb(context: *mut c_void, err: *mut weaveffi_error, result: i64) { - let tx = unsafe { &*(context as *const mpsc::Sender<(i32, i64)>) }; + let tx = unsafe { adopt_ctx::<(i32, i64)>(context) }; let code = if err.is_null() { 0 } else { @@ -813,10 +823,9 @@ mod tests { } let token = abi::cancel_token_create(); abi::cancel_token_cancel(token); - weaveffi_kv_Store_compact_async(s, token, cb, tx_ptr as *mut c_void); + weaveffi_kv_Store_compact_async(s, token, cb, ctx_for(&tx)); let (code, reclaimed) = rx.recv_timeout(Duration::from_secs(5)).unwrap(); - unsafe { drop(Box::from_raw(tx_ptr)) }; assert_eq!(code, 1004, "a cancelled compact reports KvError::IoError"); assert_eq!(reclaimed, 0); abi::cancel_token_destroy(token);