diff --git a/src/runtime/napi/napi_body.rs b/src/runtime/napi/napi_body.rs index bb0506b0b16d..456bb84a75e9 100644 --- a/src/runtime/napi/napi_body.rs +++ b/src/runtime/napi/napi_body.rs @@ -2410,6 +2410,14 @@ pub(crate) struct ThreadSafeFunction { pub callback: TsfnCallback, pub(crate) dispatch_state: AtomicU8, // DispatchState + /// JS-thread only: depth of live (possibly nested) `on_dispatch` frames. + pub(crate) dispatch_depth: u32, + /// JS-thread only: `Closed` was observed while frames or tasks still + /// reference this object; whoever is last out frees it. + pub(crate) pending_destroy: bool, + /// Queued event-loop tasks targeting `on_dispatch`; atomic because addon + /// threads schedule them. + pub(crate) inflight_dispatch_tasks: AtomicU32, pub(crate) blocking_condvar: Condvar, pub(crate) closing: AtomicU8, // ClosingState /// Written under `lock` by `env_teardown` on the JS thread. Every path @@ -2506,7 +2514,15 @@ impl ThreadSafeFunction { #[allow(clippy::not_unsafe_ptr_arg_deref)] pub(crate) fn on_dispatch(this: *mut ThreadSafeFunction) { // SAFETY: `this` is a live heap allocation owned by the event loop - // dispatch; `env_dead` is atomic so a shared reborrow suffices. + // dispatch; destroy is deferred while `dispatch_depth > 0` or tasks + // remain in flight. No `&mut *this` is held across `dispatch_one`, + // which runs user JS that can re-enter this dispatch. + let inflight = unsafe { + (*this) + .inflight_dispatch_tasks + .fetch_sub(1, Ordering::SeqCst) + } - 1; + // SAFETY: as above. if unsafe { (*this).env_dead.load(Ordering::SeqCst) } { // `env_teardown` already released everything and owns the free // decision. The loop this task came from is being destroyed. @@ -2514,6 +2530,14 @@ impl ThreadSafeFunction { } // SAFETY: as above. if unsafe { (*this).closing.load(Ordering::SeqCst) } == ClosingState::Closed as u8 { + // SAFETY: as above. + if unsafe { (*this).dispatch_depth } > 0 || inflight > 0 { + // An outer frame or queued task still references `*this`; + // whoever is last out frees it. + // SAFETY: as above. + unsafe { (*this).pending_destroy = true }; + return; + } // Finalize the ThreadSafeFunction. // SAFETY: `this` is the live heap allocation we own; closed state guarantees no other thread will touch it. unsafe { ThreadSafeFunction::destroy(this) }; @@ -2521,6 +2545,8 @@ impl ThreadSafeFunction { } let mut is_first = true; + // SAFETY: as above. + unsafe { (*this).dispatch_depth += 1 }; // Run the tasks. loop { @@ -2530,9 +2556,8 @@ impl ThreadSafeFunction { .dispatch_state .store(DispatchState::Running as u8, Ordering::SeqCst) }; - // SAFETY: as above. `dispatch_one` runs JS that can re-enter other - // TSFN entry points, so the exclusive borrow is scoped to this call. - if unsafe { (*this).dispatch_one(is_first) } { + // SAFETY: as above. + if unsafe { Self::dispatch_one(this, is_first) } { is_first = false; // SAFETY: as above. unsafe { @@ -2541,16 +2566,8 @@ impl ThreadSafeFunction { .store(DispatchState::Pending as u8, Ordering::SeqCst) }; } else { - // We're done running tasks, for now. Transition Running → Idle - // via CAS instead of an unconditional store: between - // dispatch_one() observing an empty queue (and dropping the - // lock) and this point, another thread may have enqueued an - // item and called schedule_dispatch(). That swap() saw - // Running, so it intentionally did *not* schedule a new - // concurrent task — it relies on this loop to pick the item - // up. If we blindly stored Idle we'd overwrite that Pending - // and the callback would be dropped (flaky lost-wakeup under - // load). On CAS failure, loop and re-drain. + // CAS, not a store: a concurrent schedule_dispatch() may have + // set Pending, and storing Idle over it would drop a wakeup. // SAFETY: as above. if unsafe { (*this).dispatch_state.compare_exchange( @@ -2568,6 +2585,19 @@ impl ThreadSafeFunction { } } + // SAFETY: as above. + let destroy_now = unsafe { + (*this).dispatch_depth -= 1; + (*this).dispatch_depth == 0 + && (*this).pending_destroy + && (*this).inflight_dispatch_tasks.load(Ordering::SeqCst) == 0 + }; + if destroy_now { + // SAFETY: outermost frame, no other task references `*this`. + unsafe { ThreadSafeFunction::destroy(this) }; + return; + } + // Node sets a maximum number of runs per ThreadSafeFunction to 1,000. // We don't set a max. I would like to see an issue caused by not // setting a max before we do set a max. It is better for performance to @@ -2580,10 +2610,9 @@ impl ThreadSafeFunction { /// The creating VM's event loop, or `None` once its env has been torn down. /// - /// JS-thread only. Its callers (`call`, `maybe_queue_finalizer`) run from - /// the loop's own dispatch, so no other `&mut EventLoop` is live. Paths - /// reachable from an addon thread must use the shared `&EventLoop` that - /// `BackRef` derefs to, never this. + /// JS-thread only, called from the loop's own dispatch, so no other + /// `&mut EventLoop` is live. Paths reachable from an addon thread must use + /// the shared `&EventLoop` that `BackRef` derefs to, never this. #[inline] fn loop_mut(&mut self) -> Option<&mut EventLoop> { let back_ref = self.event_loop.as_mut()?; @@ -2603,8 +2632,11 @@ impl ThreadSafeFunction { self.callback = TsfnCallback::Js(StrongOptional::empty()); self.poll_ref.disable(); let self_ptr: *mut Self = self; + // The finalize task targets `on_dispatch` too. + let _ = self.inflight_dispatch_tasks.fetch_add(1, Ordering::SeqCst); let Some(loop_) = self.loop_mut() else { // env torn down: `env_teardown` owns the finalize + free. + let _ = self.inflight_dispatch_tasks.fetch_sub(1, Ordering::SeqCst); return; }; loop_.enqueue_task(Task::init(self_ptr)); @@ -2617,48 +2649,85 @@ impl ThreadSafeFunction { } } - pub(crate) fn dispatch_one(&mut self, is_first: bool) -> bool { - let mut queue_finalizer_after_call = false; - let task = 'brk: { - // `MutexGuard` holds the lock by raw pointer, so it does not borrow - // `*self` across the `&mut self` calls below. - let _g = self.lock.lock_guard(); - let was_blocked = self.queue.is_blocked(); - let Some(t) = self.queue.data.read_item() else { - // When there are no tasks and the number of threads that have - // references reaches zero, we prepare to finalize the - // ThreadSafeFunction. - if self.thread_count.load(Ordering::SeqCst) == 0 { - if self.queue.max_queue_size > 0 { - self.blocking_condvar.signal(); - } - self.maybe_queue_finalizer(); - } - return false; - }; - - if self.queue.count.fetch_sub(1, Ordering::SeqCst) == 1 - && self.thread_count.load(Ordering::SeqCst) == 0 - { - self.closing - .store(ClosingState::Closing as u8, Ordering::SeqCst); + /// Dequeues one item under the lock, running the empty-queue finalize + /// check and scheduling the backup dispatch as needed. Never enters JS. + fn take_one_locked(&mut self, queue_finalizer_after_call: &mut bool) -> Option<*mut c_void> { + let _g = self.lock.lock_guard(); + let was_blocked = self.queue.is_blocked(); + let Some(t) = self.queue.data.read_item() else { + // When there are no tasks and the number of threads that have + // references reaches zero, we prepare to finalize the + // ThreadSafeFunction. + if self.thread_count.load(Ordering::SeqCst) == 0 { if self.queue.max_queue_size > 0 { self.blocking_condvar.signal(); } - queue_finalizer_after_call = true; - } else if was_blocked && !self.queue.is_blocked() { + self.maybe_queue_finalizer(); + } + return None; + }; + + let prev_count = self.queue.count.fetch_sub(1, Ordering::SeqCst); + if prev_count == 1 && self.thread_count.load(Ordering::SeqCst) == 0 { + self.closing + .store(ClosingState::Closing as u8, Ordering::SeqCst); + if self.queue.max_queue_size > 0 { self.blocking_condvar.signal(); } + *queue_finalizer_after_call = true; + } else if was_blocked && !self.queue.is_blocked() { + self.blocking_condvar.signal(); + } - break 'brk t; - }; + if prev_count > 1 && self.inflight_dispatch_tasks.load(Ordering::SeqCst) == 0 { + // `call` can block in a nested event loop (#36828); one backup + // dispatch keeps the queued-behind items reachable. + self.schedule_dispatch(); + } - if self.call(task, is_first).is_err() { - return false; + Some(t) + } + + /// Runs one queued call. `this` stays raw: the microtask drain and `call` + /// enter user JS, which can re-enter this function's dispatch through a + /// nested event loop. + /// + /// # Safety + /// `this` is a live threadsafe function on the JS thread. + unsafe fn dispatch_one(this: *mut Self, is_first: bool) -> bool { + // Drain microtasks between callbacks (node#38506) BEFORE dequeuing: + // if the drain blocks in a nested event loop, the next item is still + // in the queue for a nested dispatch to deliver, in push order. + if !is_first { + let loop_: *mut EventLoop = { + // SAFETY: scoped reborrow. + match unsafe { (*this).event_loop.as_mut() } { + // SAFETY: BackRef invariant while `Some`; JS thread, + // outside tick(). + Some(back_ref) => unsafe { back_ref.get_mut() }, + None => return false, + } + }; + // SAFETY: the per-thread event loop outlives this call. + if unsafe { (*loop_).drain_microtasks() }.is_err() { + return false; + } } + let mut queue_finalizer_after_call = false; + // SAFETY: call-scoped reborrow; `take_one_locked` never enters JS. + let Some(task) = (unsafe { (*this).take_one_locked(&mut queue_finalizer_after_call) }) + else { + return false; + }; + + // No borrow of `*this` is live while user JS runs. + // SAFETY: per fn contract. + unsafe { Self::call(this, task) }; + if queue_finalizer_after_call { - self.maybe_queue_finalizer(); + // SAFETY: call-scoped reborrow; `this` is still live (destroy deferred). + unsafe { (*this).maybe_queue_finalizer() }; } // An item was dequeued: keep on_dispatch looping so remaining queued @@ -2666,52 +2735,72 @@ impl ThreadSafeFunction { true } - /// This function can be called multiple times in one tick of the event loop. - /// See: https://github.com/nodejs/node/pull/38506 - /// In that case, we need to drain microtasks. - fn call(&mut self, task: *mut c_void, is_first: bool) -> Result<(), bun_jsc::JsTerminated> { - let Some(env) = self.env.as_ref().map(NapiEnvRef::get) else { + /// Invokes the callback for one item. `this` stays raw: the callback runs + /// user JS, which can re-enter this function's dispatch through a nested + /// event loop. + /// + /// # Safety + /// `this` is a live threadsafe function on the JS thread. + unsafe fn call(this: *mut Self, task: *mut c_void) { + // SAFETY: scoped reborrow; `env` is a raw pointer copy. + let Some(env) = (unsafe { &(*this).env }).as_ref().map(NapiEnvRef::get) else { // env torn down; nothing to call into. - return Ok(()); + return; }; - if !is_first { - let Some(loop_) = self.loop_mut() else { - return Ok(()); - }; - loop_.drain_microtasks()?; + + // Copy the callback target out so no borrow of `*this` is live while + // user code runs. + enum Target { + Js(JSValue), + C( + napi_threadsafe_function_call_js, + Option, + *mut c_void, + ), } + // SAFETY: scoped reborrow. + let target = match unsafe { &(*this).callback } { + TsfnCallback::Js(strong) => Target::Js(strong.get().unwrap_or(JSValue::UNDEFINED)), + TsfnCallback::C { + js: cb_js, + napi_threadsafe_function_call_js, + } => Target::C( + *napi_threadsafe_function_call_js, + cb_js.get(), + // SAFETY: plain field read, same reborrow scope. + unsafe { (*this).ctx }, + ), + }; + // SAFETY: env is valid while the TSF is live. let global_object = unsafe { &*env }.to_js(); - let _dispatch = self.tracker.dispatch(global_object); + // SAFETY: scoped reborrow; `tracker` is `Copy`, the guard borrows only + // `global_object`. + let _dispatch = unsafe { (*this).tracker }.dispatch(global_object); - match &self.callback { - TsfnCallback::Js(strong) => { - let js: JSValue = strong.get().unwrap_or(JSValue::UNDEFINED); + match target { + Target::Js(js) => { if js.is_empty_or_undefined_or_null() { - return Ok(()); + return; } let _ = js .call(global_object, JSValue::UNDEFINED, &[]) .map_err(|err| global_object.report_active_exception_as_unhandled(err)); } - TsfnCallback::C { - js: cb_js, - napi_threadsafe_function_call_js, - } => { - // SAFETY: `env` is held alive by `self.env` (`NapiEnvRef`) for the TSF's lifetime. + Target::C(call_js, cb_js, ctx) => { + // SAFETY: `env` is held alive by `(*this).env` (`NapiEnvRef`) for the TSF's lifetime. let env_ref = unsafe { &*env }; let _hs = NapiHandleScope::open_scoped(env_ref); // No func at creation => null js_callback (Node), not encoded undefined. - let js = match cb_js.get() { + let js = match cb_js { Some(v) => napi_value::create(env_ref, v), None => napi_value(0), }; - napi_threadsafe_function_call_js(env, js, self.ctx, task); + call_js(env, js, ctx, task); } } - Ok(()) } /// Runs on an addon thread. A call that reports `napi_closing` consumes the @@ -2773,29 +2862,25 @@ impl ThreadSafeFunction { (NapiStatus::ok as napi_status, false) } - /// Caller must hold `lock`. Reached from addon threads (`enqueue`, - /// `release_locked`), so it may only take a shared `&EventLoop`: the JS - /// thread can be inside `tick()` with its own `&mut` at the same time. + /// Caller must hold `lock`. Reachable from addon threads, so it may only + /// take a shared `&EventLoop`: the JS thread can be inside `tick()` with + /// its own `&mut` at the same time. fn schedule_dispatch(&mut self) { - let prev = self + let _ = self .dispatch_state .swap(DispatchState::Pending as u8, Ordering::SeqCst); - match prev { - x if x == DispatchState::Idle as u8 => { - let self_ptr: *mut Self = self; - let Some(event_loop) = self.event_loop.as_ref() else { - // env torn down: the loop is gone, nothing to schedule onto. - return; - }; - event_loop.enqueue_task_concurrent(ConcurrentTask::create_from(self_ptr)); - } - x if x == DispatchState::Running as u8 => { - // it will check if it has more work to do - } - _ => { - // we've already scheduled it to run - } + if self.inflight_dispatch_tasks.load(Ordering::SeqCst) > 0 { + // A queued task will pick the item up, even if the running loop + // is blocked in a nested event loop (#36828). + return; } + let self_ptr: *mut Self = self; + let Some(event_loop) = self.event_loop.as_ref() else { + // env torn down: the loop is gone, nothing to schedule onto. + return; + }; + let _ = self.inflight_dispatch_tasks.fetch_add(1, Ordering::SeqCst); + event_loop.enqueue_task_concurrent(ConcurrentTask::create_from(self_ptr)); } /// Consumes and frees a heap-allocated ThreadSafeFunction (allocated by `new`). @@ -3082,6 +3167,9 @@ extern "C" fn napi_create_threadsafe_function( has_queued_finalizer: false, lock: Mutex::new(), dispatch_state: AtomicU8::new(DispatchState::Idle as u8), + dispatch_depth: 0, + pending_destroy: false, + inflight_dispatch_tasks: AtomicU32::new(0), blocking_condvar: Condvar::default(), closing: AtomicU8::new(ClosingState::NotClosing as u8), env_dead: AtomicBool::new(false), diff --git a/test/napi/napi-app/binding.gyp b/test/napi/napi-app/binding.gyp index 94f07ed09603..6dc842aabe65 100644 --- a/test/napi/napi-app/binding.gyp +++ b/test/napi/napi-app/binding.gyp @@ -286,5 +286,18 @@ "NODE_API_EXPERIMENTAL_NOGC_ENV_OPT_OUT=1", ], }, + { + "target_name": "tsfn_nested_wait_addon", + "cflags!": ["-fno-exceptions"], + "cflags_cc!": ["-fno-exceptions"], + "sources": ["tsfn_nested_wait_addon.cpp"], + "include_dirs": [" { + const done = Promise.withResolvers(); + const blocker = Promise.withResolvers(); + const order: number[] = []; + addon.startConcurrent((tag: number) => { + order.push(tag); + if (tag === 1) { + addon.signalBlocked(); + // blocks inside this dispatch until call 2 is dispatched + expect(blocker.promise).resolves.toBe(2); + order.push(3); + done.resolve(); + } else { + blocker.resolve(tag); + } + }); + await done.promise; + expect(order).toEqual([1, 2, 3]); +}); + +test("calls queued behind a callback that blocks in a nested wait are dispatched", async () => { + const done = Promise.withResolvers(); + const blocker = Promise.withResolvers(); + const order: number[] = []; + // Both calls are already queued before the first dispatch runs. + addon.startQueued((tag: number) => { + order.push(tag); + if (tag === 1) { + expect(blocker.promise).resolves.toBe(2); + order.push(3); + done.resolve(); + } else { + blocker.resolve(tag); + } + }); + await done.promise; + expect(order).toEqual([1, 2, 3]); +}); + +test("calls queued behind a microtask that blocks in a nested wait are dispatched in order", async () => { + const done = Promise.withResolvers(); + const blocker = Promise.withResolvers(); + const order: number[] = []; + addon.startQueued((tag: number) => { + order.push(tag); + if (tag === 1) { + // The block happens in the microtask drained between the two queued + // calls, not in the callback itself. + queueMicrotask(() => { + expect(blocker.promise).resolves.toBe(2); + order.push(3); + done.resolve(); + }); + } else { + blocker.resolve(tag); + } + }); + await done.promise; + expect(order).toEqual([1, 2, 3]); +}); diff --git a/test/napi/napi-app/tsfn_nested_wait_addon.cpp b/test/napi/napi-app/tsfn_nested_wait_addon.cpp new file mode 100644 index 000000000000..462f47252f48 --- /dev/null +++ b/test/napi/napi-app/tsfn_nested_wait_addon.cpp @@ -0,0 +1,151 @@ +// Regression test for oven-sh/bun#36828: threadsafe function calls must keep +// dispatching while a callback blocks in a nested event loop +// (expect(promise).resolves under bun:test). + +#include +#include + +#include +#include +#include + +#define NODE_API_CALL(env, call) \ + do { \ + napi_status status = (call); \ + if (status != napi_ok) { \ + const napi_extended_error_info *error_info = NULL; \ + napi_get_last_error_info((env), &error_info); \ + const char *err_message = error_info->error_message; \ + bool is_pending; \ + napi_is_exception_pending((env), &is_pending); \ + if (!is_pending) { \ + const char *message = \ + (err_message == NULL) ? "empty error message" : err_message; \ + napi_throw_error((env), NULL, message); \ + } \ + return NULL; \ + } \ + } while (0) + +static std::atomic blocked_signal{false}; + +// Invokes the JS callback with the integer tag passed to +// napi_call_threadsafe_function. +static void call_js(napi_env env, napi_value js_callback, void *context, + void *data) { + (void)context; + if (env == nullptr || js_callback == nullptr) { + return; + } + napi_value tag; + if (napi_create_int32(env, (int32_t)(intptr_t)data, &tag) != napi_ok) { + return; + } + napi_value recv; + if (napi_get_undefined(env, &recv) != napi_ok) { + return; + } + napi_value result; + // A pending exception (e.g. a failed expect() in the callback) is + // surfaced by the fixture's test runner; nothing to do here. + (void)napi_call_function(env, recv, js_callback, 1, &tag, &result); +} + +static napi_status make_tsfn(napi_env env, napi_value js_callback, + const char *name_str, + napi_threadsafe_function *out) { + napi_value name; + napi_status status = + napi_create_string_utf8(env, name_str, NAPI_AUTO_LENGTH, &name); + if (status != napi_ok) { + return status; + } + status = napi_create_threadsafe_function( + env, js_callback, /* async_resource */ NULL, name, + /* max_queue_size (unlimited) */ 0, /* initial_thread_count */ 1, + /* thread_finalize_data */ NULL, /* thread_finalize_cb */ NULL, + /* context */ NULL, call_js, out); + if (status != napi_ok) { + return status; + } + // Unreferenced so a deadlocked run still exits once its test times out, + // instead of the pending call keeping the event loop alive forever. + return napi_unref_threadsafe_function(env, *out); +} + +// Pushes call 1, then once the fixture signals that call 1's callback is +// blocked in its nested wait, pushes call 2 from the addon thread. +static napi_value start_concurrent(napi_env env, napi_callback_info info) { + size_t argc = 1; + napi_value args[1]; + NODE_API_CALL(env, napi_get_cb_info(env, info, &argc, args, NULL, NULL)); + + blocked_signal.store(false); + + napi_threadsafe_function tsfn; + NODE_API_CALL(env, make_tsfn(env, args[0], "tsfn_nested_wait_concurrent", &tsfn)); + + std::thread([tsfn] { + napi_call_threadsafe_function(tsfn, (void *)1, napi_tsfn_nonblocking); + // Bounded wait so a broken fixture cannot spin this thread forever. + for (int i = 0; i < 30000 && !blocked_signal.load(); i++) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + napi_call_threadsafe_function(tsfn, (void *)2, napi_tsfn_nonblocking); + napi_release_threadsafe_function(tsfn, napi_tsfn_release); + }).detach(); + + return NULL; +} + +// Pushes calls 1 and 2 before any dispatch can run, so call 2 is already +// queued behind call 1 when call 1's callback blocks. +static napi_value start_queued(napi_env env, napi_callback_info info) { + size_t argc = 1; + napi_value args[1]; + NODE_API_CALL(env, napi_get_cb_info(env, info, &argc, args, NULL, NULL)); + + napi_threadsafe_function tsfn; + NODE_API_CALL(env, make_tsfn(env, args[0], "tsfn_nested_wait_queued", &tsfn)); + + NODE_API_CALL(env, + napi_call_threadsafe_function(tsfn, (void *)1, napi_tsfn_nonblocking)); + NODE_API_CALL(env, + napi_call_threadsafe_function(tsfn, (void *)2, napi_tsfn_nonblocking)); + NODE_API_CALL(env, napi_release_threadsafe_function(tsfn, napi_tsfn_release)); + + return NULL; +} + +// Called by the fixture from inside call 1's callback, right before it starts +// the nested wait. +static napi_value signal_blocked(napi_env env, napi_callback_info info) { + (void)env; + (void)info; + blocked_signal.store(true); + return NULL; +} + +static napi_value init(napi_env env, napi_value exports) { + napi_value fn; + + NODE_API_CALL(env, napi_create_function(env, "startConcurrent", + NAPI_AUTO_LENGTH, start_concurrent, + NULL, &fn)); + NODE_API_CALL(env, + napi_set_named_property(env, exports, "startConcurrent", fn)); + + NODE_API_CALL(env, napi_create_function(env, "startQueued", NAPI_AUTO_LENGTH, + start_queued, NULL, &fn)); + NODE_API_CALL(env, napi_set_named_property(env, exports, "startQueued", fn)); + + NODE_API_CALL(env, napi_create_function(env, "signalBlocked", + NAPI_AUTO_LENGTH, signal_blocked, + NULL, &fn)); + NODE_API_CALL(env, + napi_set_named_property(env, exports, "signalBlocked", fn)); + + return exports; +} + +NAPI_MODULE(NODE_GYP_MODULE_NAME, init) diff --git a/test/napi/tsfn-nested-wait.test.ts b/test/napi/tsfn-nested-wait.test.ts new file mode 100644 index 000000000000..575d18d1a222 --- /dev/null +++ b/test/napi/tsfn-nested-wait.test.ts @@ -0,0 +1,87 @@ +// Regression test for oven-sh/bun#36828: threadsafe function calls deadlocked +// under `bun test` when a callback blocked in a nested event loop +// (expect(promise).resolves / .rejects waits that way). +import { spawn, spawnSync } from "bun"; +import { beforeAll, expect, it } from "bun:test"; +import { existsSync, statSync } from "fs"; +import { bunEnv, bunExe, canBuildNodeAddons } from "harness"; +import { join } from "path"; + +const napiAppDir = join(__dirname, "napi-app"); +const addonName = "tsfn_nested_wait_addon"; +const addonPath = join(napiAppDir, `build/Debug/${addonName}.node`); + +beforeAll(() => { + if (!canBuildNodeAddons()) return; + // Build the native addons in napi-app, but only if the one this test needs + // is missing or older than its inputs (napi.test.ts or a previous run + // usually has built it already). The addon doesn't link against bun, so an + // existing binary stays valid across bun builds; skipping the install avoids + // re-running the node-gyp rebuild, which is slow and occasionally flaky + // under resource pressure. + if (existsSync(addonPath)) { + const built = statSync(addonPath).mtimeMs; + const inputs = [join(napiAppDir, "tsfn_nested_wait_addon.cpp"), join(napiAppDir, "binding.gyp")]; + if (inputs.every(f => statSync(f).mtimeMs < built)) { + return; + } + } + if (!existsSync(join(napiAppDir, "node_modules/node-gyp"))) { + spawnSync({ + cmd: [bunExe(), "install", "--ignore-scripts", "--verbose"], + cwd: napiAppDir, + stderr: "inherit", + env: bunEnv, + stdout: "inherit", + stdin: "inherit", + }); + } + spawnSync({ + // Same invocation as napi-app's install script, minus `clean` and scoped + // to one target (node-gyp forwards trailing args as make/msbuild targets). + cmd: [bunExe(), "--bun", "node-gyp", "configure", "build", "--debug", "-j", "max", addonName], + cwd: napiAppDir, + stderr: "inherit", + env: bunEnv, + stdout: "inherit", + stdin: "inherit", + }); + if (existsSync(addonPath)) { + return; + } + // Fallback: the full install + rebuild of everything, retried once. + for (let attempt = 0; ; attempt++) { + const install = spawnSync({ + cmd: [bunExe(), "install", "--verbose"], + cwd: napiAppDir, + stderr: "inherit", + env: bunEnv, + stdout: "inherit", + stdin: "inherit", + }); + if (install.success && existsSync(addonPath)) { + return; + } + if (attempt >= 1) { + throw new Error("building napi-app addons failed"); + } + } +}, 300_000); + +it.skipIf(!canBuildNodeAddons())( + "threadsafe function calls are dispatched while a callback blocks in a nested promise wait", + async () => { + await using proc = spawn({ + cmd: [bunExe(), "test", join(napiAppDir, "tsfn-nested-wait.fixture.ts")], + cwd: napiAppDir, + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toContain("3 pass"); + expect(stderr).toContain("0 fail"); + expect(exitCode).toBe(0); + }, + 60_000, +);