From 94c84ae7b0d0606a0b95fa4afaf15d73e4536998 Mon Sep 17 00:00:00 2001 From: Owen Carey <37121709+owenthcarey@users.noreply.github.com> Date: Wed, 16 Sep 2026 21:50:45 -0700 Subject: [PATCH] test: give each async launch an owned callback context The async tests passed the completion callback a raw pointer to a boxed mpsc::Sender, then freed that box on the test thread right after recv returned. recv unblocks as soon as the message is queued, before the producer thread has finished send, so the test could free the Sender and then the whole channel out from under an in-flight send. The window is tiny and load dependent, which is why it surfaced as an intermittent SIGSEGV in the weaveffi runtime tests on macOS CI and nowhere else. Every launch now gets its own boxed clone of the sender and the callback adopts it with Box::from_raw, so the channel is torn down by whichever side releases last and no borrowed reference outlives the callback. This is the ownership the ABI's "consumer owns context" rule intends, and it replaces the deliberate leak async-demo used to sidestep the same race. The generated launchers were already correct: they carry context as an opaque usize and never touch it after the callback fires. --- crates/weaveffi/tests/runtime.rs | 74 +++++++++++++++++--------------- samples/async-demo/src/lib.rs | 70 +++++++++++------------------- samples/kvstore/src/lib.rs | 25 +++++++---- 3 files changed, 83 insertions(+), 86 deletions(-) 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);