diff --git a/src/install/patch_install.rs b/src/install/patch_install.rs index 2b5addb0149b..f9359868daa6 100644 --- a/src/install/patch_install.rs +++ b/src/install/patch_install.rs @@ -141,24 +141,35 @@ impl PatchTask { /// # Safety /// Only invoked by `ThreadPool` via the `callback` fn-pointer registered in /// `new_calc_patch_hash` / `new_apply_patch_hash`. `task` must be live and - /// point at the `task` field of a heap-allocated `PatchTask`, with the pool - /// granting exclusive access for the duration of the call. + /// point at the `task` field of a heap-allocated `PatchTask`, which this + /// thread owns until it pushes the task back to the main thread. pub(crate) unsafe fn run_from_thread_pool(task: *mut ThreadPoolTask) { // SAFETY: thread-pool callback contract — `task` points to the `task` // field of a live `PatchTask` (set at construction); the pool runs // each task at most once with exclusive access for the call. - let patch_task = unsafe { &mut *PatchTask::from_task_ptr(task) }; - patch_task.run_from_thread_pool_impl(); + let this = unsafe { PatchTask::from_task_ptr(task) }; + // SAFETY: as above; the `&mut` ends with its statement. The main thread + // frees the task as soon as the push below lands. + let mgr = unsafe { + (*this).run_from_thread_pool_impl(); + (*this).manager.as_ptr() + }; + // SAFETY: `this` is non-null (as above). `mgr` is a long-lived BACKREF; + // the lock-free queue and the wake atomics are all this thread touches. + unsafe { + (*mgr) + .patch_task_queue + .push(core::ptr::NonNull::new_unchecked(this)); + PackageManager::wake_raw(mgr); + } } - pub(crate) fn run_from_thread_pool_impl(&mut self) { + fn run_from_thread_pool_impl(&mut self) { bun_output::scoped_log!( InstallPatch, "runFromThreadPoolImpl {}", <&'static str>::from(&self.callback) ); - // There are no early returns in the body, so the ordering - // (body → push → wake) is inlined below. match &mut self.callback { Callback::CalcHash(_) => { let result = self.calc_hash(); @@ -173,17 +184,6 @@ impl PatchTask { self.apply().expect("OOM"); } } - let mgr = self.manager.as_ptr(); - // SAFETY: `self.manager` is a long-lived BACKREF; - // the worker thread only touches the lock-free `patch_task_queue` and the - // event-loop wake atomics, neither of which alias data the main thread - // holds an exclusive borrow on. - unsafe { - (*mgr) - .patch_task_queue - .push(core::ptr::NonNull::from(&mut *self)); - PackageManager::wake_raw(mgr); - } } pub(crate) fn run_from_main_thread( diff --git a/src/jsc/RuntimeTranspilerStore.rs b/src/jsc/RuntimeTranspilerStore.rs index 609a8c9b87f1..b5fcb85e7474 100644 --- a/src/jsc/RuntimeTranspilerStore.rs +++ b/src/jsc/RuntimeTranspilerStore.rs @@ -240,20 +240,26 @@ impl RuntimeTranspilerStore { /// queued after the last tick, or posted after `close()` began — release /// their source, log and module promise here instead of running. pub fn release_queued_jobs_for_teardown(&mut self) { - let batch = self.queue.pop_batch(); - let mut iter = batch.iterator(); - loop { - let job = iter.next(); - if job.is_null() { - break; - } - // SAFETY: a live job popped from the intrusive queue; this thread - // owns it now (its worker-thread part finished before `close()`). + let mut iter = self.queue.pop_batch().iterator(); + let first = iter.next(); + self.release_unrun_jobs(first, &mut iter); + } + + /// Releases `job` and the rest of `iter` without running their completions. + fn release_unrun_jobs( + &self, + mut job: *mut TranspilerJob, + iter: &mut unbounded_queue::BatchIterator, + ) { + while !job.is_null() { + // SAFETY: a live job popped from the queue, owned by this thread since + // the pool pushed it; `iter` has already moved past it. unsafe { (*job).promise.deinit(); (*job).reset_for_pool(); self.store.put(job); } + job = iter.next(); } } @@ -272,8 +278,9 @@ impl RuntimeTranspilerStore { return; } // we run just one job first to see if there are more - // SAFETY: `first` is a live job popped from the intrusive queue. - if let Err(err) = unsafe { (*first).run_from_js_thread() } { + // SAFETY: `first` is a live job popped from the intrusive queue; the + // batch iterator has already moved past it. + if let Err(err) = unsafe { TranspilerJob::run_from_js_thread(first) } { global.report_uncaught_exception_from_error(err); } loop { @@ -286,10 +293,12 @@ impl RuntimeTranspilerStore { if unsafe { (*event_loop.as_ptr()).drain_microtasks_with_global(global, jsc_vm) } .is_err() { + // Terminated; the rest of the batch is off the queue, so only we can release it. + self.release_unrun_jobs(job, &mut iter); return; } - // SAFETY: `job` is a live job popped from the intrusive queue. - if let Err(err) = unsafe { (*job).run_from_js_thread() } { + // SAFETY: as for `first`. + if let Err(err) = unsafe { TranspilerJob::run_from_js_thread(job) } { global.report_uncaught_exception_from_error(err); } } @@ -403,9 +412,8 @@ pub struct TranspilerJob { // raw pointers/BackRefs are used (BACKREF — VM owns the // store and outlives every job). pub(crate) vm: *mut VirtualMachine, - /// The pool thread runs this job under `loop_handle.borrow()`: the job's - /// own slot, the transpiler it copies and the store queue it pushes to are - /// all VM-owned, and the VM's teardown waits for the borrow to end. + /// The pool thread transpiles under `borrow_if_running()` and is counted as + /// embedded work from `schedule` until it has pushed the slot back. pub(crate) loop_handle: crate::LoopHandle, pub global_this: BackRef, pub(crate) fetcher: Fetcher, @@ -447,6 +455,18 @@ impl Fetcher { } } +/// A finished job's result, moved out of its slot so the slot can go back to +/// the store before `AsyncModule::fulfill` runs. +struct Completion { + promise: JSValue, + global_this: BackRef, + specifier: String, + referrer: String, + log: bun_ast::Log, + resolved_source: OwnedResolvedSource, + parse_error: Option, +} + /// Per-worker output buffer. The printer is the **only** state /// retained across `run()` calls — its backing `Vec` is genuinely worth /// reusing (capped at 512 K / 2 M below). The parse arena and AST memory @@ -484,8 +504,9 @@ fn tls_get_or_leak( impl TranspilerJob { /// Kept as a private inherent fn (not `impl Drop`) because the - /// slot is recycled into the HiveArray via `store.put(this)`. Only caller is - /// `run_from_js_thread`. + /// slot is recycled into the HiveArray via `store.put(this)`. Called right + /// before that `put` by `take_completion` and + /// `release_queued_jobs_for_teardown`. /// /// Note: `HiveArrayFallback::put` runs `drop_in_place` on the slot (see /// hive_array.rs note), so the Drop-carrying fields — `OwnedString` ×2, @@ -514,18 +535,21 @@ impl TranspilerJob { // replacement a second time). } - fn dispatch_to_main_thread(&mut self) { - let vm = self.vm; - let loop_handle = self.loop_handle.clone(); + /// Hands the job to the JS thread, which recycles or frees the slot as soon + /// as the push lands: no reference to it (a `&mut self` included, it stays + /// protected until the call returns) may be live across the push. + /// + /// # Safety + /// `this` is the live slot this pool thread owns; it is not touched afterwards. + unsafe fn dispatch_to_main_thread(this: *mut Self) { + // SAFETY: fn contract; both accesses end before the push. + let (vm, loop_handle) = unsafe { ((*this).vm, (*this).loop_handle.clone()) }; // SAFETY: vm outlives the job (BACKREF — VM owns the store). let transpiler_store: *mut RuntimeTranspilerStore = unsafe { ptr::addr_of_mut!((*vm).transpiler_store) }; - let job = NonNull::from(&mut *self); - // SAFETY: queue is concurrent-safe (UnboundedQueue uses atomics). - unsafe { (*transpiler_store).queue.push(job) }; - // Another thread may free `self` at any time after .push, so we cannot use it any more - // (the handle was cloned out above for exactly this reason). The VM - // waits for embedded work before closing its handle, so this is queued. + // SAFETY: `this` is non-null (fn contract); the queue is concurrent-safe. + unsafe { (*transpiler_store).queue.push(NonNull::new_unchecked(this)) }; + // The VM waits for embedded work before closing its handle, so this is queued. let crate::vm_handle::Posted::Queued = loop_handle.post_task(ConcurrentTask::create_from(transpiler_store)) else { @@ -533,21 +557,49 @@ impl TranspilerJob { }; } - fn run_from_js_thread(&mut self) -> JsResult<()> { - let vm = self.vm; + /// The slot is recycled or freed before `fulfill` runs script, so, as in + /// `dispatch_to_main_thread`, no reference to it may be live at that point. + /// + /// # Safety + /// `this` is a live job popped from the store queue; it is not touched afterwards. + unsafe fn run_from_js_thread(this: *mut Self) -> JsResult<()> { + // SAFETY: fn contract; both accesses end before the put. + let (vm, completion) = unsafe { ((*this).vm, (*this).take_completion()) }; + // SAFETY: `vm` outlives the job; `this` came from this store's `get_init`. + unsafe { (*vm).transpiler_store.store.put(this) }; + + let Completion { + promise, + global_this, + specifier, + referrer, + mut log, + resolved_source, + parse_error, + } = completion; + let mut resolved_source = resolved_source.into_ffi(); + AsyncModule::fulfill( + &global_this, + promise, + &mut resolved_source, + parse_error, + specifier, + referrer, + &mut log, + ) + } + + /// Moves the result out of the slot and resets it for `store.put()`. + fn take_completion(&mut self) -> Completion { let promise = self.promise.swap(); - // Copy the BackRef out (it is `Copy`) so the borrow of `*self` ends - // before `reset_for_pool`/`put` need `&mut *self` below; deref at the - // `fulfill` call site instead. let global_this = self.global_this; // Note: the KeepAlive takes an `EventLoopCtx` // vtable; resolve it via the `get_vm_ctx` hook (registered by `bun_runtime::init`). self.poll_ref.unref(get_vm_ctx(AllocatorType::Js)); let referrer = core::mem::take(&mut self.non_threadsafe_referrer).into_inner(); - let mut log = core::mem::replace(&mut self.log, bun_ast::Log::init()); - // Take RAII ownership out of the job; `into_ffi()` below transfers the - // +1 strings to `AsyncModule::fulfill` → C++ `Zig::ResolvedSource`. + let log = core::mem::replace(&mut self.log, bun_ast::Log::init()); + // The caller's `into_ffi()` transfers the +1 strings to `AsyncModule::fulfill`. let mut owned_resolved_source = core::mem::take(&mut self.resolved_source); let resolved_source = owned_resolved_source.as_mut(); let specifier = 'brk: { @@ -569,24 +621,15 @@ impl TranspilerJob { self.promise.deinit(); self.reset_for_pool(); - // SAFETY: vm outlives the job; transpiler_store.store.put recycles the slot. - unsafe { - (*vm) - .transpiler_store - .store - .put(std::ptr::from_mut::(self)) - }; - - let mut resolved_source = owned_resolved_source.into_ffi(); - AsyncModule::fulfill( - &global_this, + Completion { promise, - &mut resolved_source, - parse_error, + global_this, specifier, referrer, - &mut log, - ) + log, + resolved_source: owned_resolved_source, + parse_error, + } } fn schedule(&mut self) { @@ -613,16 +656,16 @@ impl TranspilerJob { // SAFETY: as above. let handle = unsafe { (*this).loop_handle.clone() }; if let Some(_vm) = handle.borrow_if_running() { - // SAFETY: live slot, exclusively ours until dispatched. + // SAFETY: live slot, exclusively ours until dispatched; the `&mut` + // ends with this statement. unsafe { (*this).run() }; - } else { - // SAFETY: as above. - unsafe { (*this).dispatch_to_main_thread() }; } - // Last touch of the slot from this thread was the dispatch. + // SAFETY: as above; this thread's last touch of the slot. + unsafe { Self::dispatch_to_main_thread(this) }; handle.embedded_work_finished(); } + /// Fills in `resolved_source` or `parse_error`; the caller dispatches. fn run(&mut self) { // Stack-local per call, bulk-freed on return. An earlier version hoisted // this to a per-worker-thread leaked `Box` (and a second @@ -640,14 +683,6 @@ impl TranspilerJob { // between calls. let arena = Arena::new(); - // `defer this.dispatchToMainThread()` — fires on every return path. - let this_ptr: *mut TranspilerJob = self; - scopeguard::defer! { - // SAFETY: `self` outlives this guard (guard drops before fn return); - // no other &mut alias is live at drop time. - unsafe { (*this_ptr).dispatch_to_main_thread() }; - } - // SAFETY contract: `vm` outlives the job (BACKREF — VM owns the store). // Note: kept as a raw pointer — never form `&mut VirtualMachine` // here. (a) the JS thread is concurrently live on the same VM, so a diff --git a/test/internal/source-lints/self-receiver-push-put.test.ts b/test/internal/source-lints/self-receiver-push-put.test.ts new file mode 100644 index 000000000000..dfb187ae9816 --- /dev/null +++ b/test/internal/source-lints/self-receiver-push-put.test.ts @@ -0,0 +1,281 @@ +import { file } from "bun"; +import { expect, test } from "bun:test"; +import { realpathSync } from "fs"; +import path from "path"; +import { globAllSources } from "../../../scripts/glob-sources.ts"; + +// A method must not push its own receiver onto a queue or put it back into its +// pool. Inside a `&mut self` method, the receiver's address +// +// queue.push(NonNull::from(&mut *self)) +// store.put(ptr::from_mut(self)) +// let job = NonNull::from(&mut *self); ... queue.push(job) +// let p: *mut Self = self; ... queue.push(NonNull::new_unchecked(p)) +// +// as the argument of a `.push(..)` or `.put(..)` is banned. +// +// Both calls give the receiver's storage away. A push is how a pool thread +// hands a finished job back to the thread that owns it (`UnboundedQueue::push`, +// then a wakeup); from the moment it lands the consumer writes to the object or +// frees it (the package manager reclaims the `Box` a patch task lives in). A +// `put` returns a slot to its hive (`HiveArrayFallback::put`), which drops it +// in place, or frees it when it was a heap spill, before returning. Either way +// the `self` of the method that made the call is still a live argument while +// that happens. A reference argument is protected for the duration of its +// call, and writing to or deallocating protected memory is UB under both +// aliasing models whether or not the method touches `self` again: Tree Borrows +// (what `bun run rust:miri` uses) reports "deallocation through is +// forbidden ... the strongly protected tag disallows deallocations" or, for the +// cross-thread writes, a data race on the protector's release, pointing at the +// receiver; Stacked Borrows reports "deallocating while item [Unique] is +// strongly protected". Codegen relies on the same guarantee: a `&mut self` +// argument is `noalias dereferenceable` for the whole call, so the compiler is +// free to re-read a field through it after the call instead of keeping the copy +// taken before; a read after the push (spelled out in the source, there) is +// what crashed the Zig version of the transpiler store (#29128). +// `TranspilerJob::dispatch_to_main_thread(&mut self)` (reached from inside +// `run(&mut self)` via a scope guard, so two protected frames deep), +// `TranspilerJob::run_from_js_thread(&mut self)` (the `put` of the same slot) +// and `PatchTask::run_from_thread_pool_impl(&mut self)` were the instances this +// was written for. +// +// The object was a raw pointer in the caller's hands before it became `self` +// (the pool recovers it from the intrusive task field, the queue pops it), so +// the fix is to keep it one: run the body through a statement-scoped reborrow +// (`(*this).run();`, `(*this).take_completion()`), read whatever comes after +// through `(*this).field` accesses that end before the call, push or put +// `this`, and never touch it again. Templates: `TranspilerJob:: +// run_from_worker_thread` / `dispatch_to_main_thread` / `run_from_js_thread` in +// src/jsc/RuntimeTranspilerStore.rs, `PatchTask::run_from_thread_pool` in +// src/install/patch_install.rs, `NetworkTask::notify` in +// src/install/NetworkTask.rs, `Task::callback` in src/install/PackageManagerTask.rs. +// +// Scope: the exclusive spellings below (`from_mut(self)`, `&mut *self`, +// `&raw mut *self`, `self as *mut _`, ...), with `self` as the receiver and +// `.push(` / `.put(` as the callee, inline or through a local of the same +// function. Outside it: the shared spellings (`from_ref(self)`, `&*self`; what +// this tree pushes from a `&self` method is a same-thread registry entry, +// quic's `ENDPOINT_REGISTRY`, whose consumer neither writes nor frees), a bare +// `self` argument (the `put(self, ..)` map inserts and `err.put(global, ..)` +// property writes, where `self` is a key or a context, not storage being given +// away), a pointer produced by a helper (`self.as_ptr()`), something the +// receiver owns (`NonNull::from(&mut self.task)`, `&raw mut self.task`), and +// reference parameters other than `self` (`fn f(this: &mut Task)` pushing +// `NonNull::from(this)`; a local reference is not protected and the push moves +// it, but a reference *parameter* pushed that way is the same bug, convert it +// on sight). Other helpers that free their argument (`Self::destroy(..)`) are +// the population self-receiver-reclaim.test.ts names as outside its scope. +// Siblings: self-receiver-reclaim.test.ts, fn-long-mut-reborrow.test.ts, +// frozen-nonnull-reborrow.test.ts. + +const root = path.resolve(import.meta.dir, "..", "..", ".."); +const rustSources = globAllSources().rust.filter(p => p.endsWith(".rs")); + +// Only scan files tracked in HEAD (a `git stash` round-trip can leave stray +// `.rs` files in the working tree; CI runs on a clean checkout). Same guard as +// dead-code-escapes.test.ts. +const tracked: Set | null = (() => { + const r = Bun.spawnSync({ + cmd: ["git", "-C", root, "ls-tree", "-r", "--name-only", "-z", "HEAD"], + stdout: "pipe", + stderr: "ignore", + }); + if (!r.success) return null; + return new Set(r.stdout.toString().split("\0").filter(Boolean)); +})(); + +// The hand-over. Anchored on the argument below, so the `Vec::push` and map +// `put` calls all over the tree only count when the argument is the receiver's +// address. `\s*` after the paren so a rustfmt-wrapped argument still matches. +const HAND_OVER = String.raw`\.\s*(?:push|put)\s*\(\s*`; + +// Optional path in front of an item (`core::ptr::NonNull`, `std::ptr::from_mut`). +const PATH = String.raw`(?:[\w:]+::)?`; +const TURBOFISH = String.raw`(?:::<[^>]*>)?`; + +// `self` as a `*mut`. The `&raw` form needs the `(?!\s*\.)` so +// `&raw mut *self.field` (something the receiver owns) stays out; `self as` +// swallows the pointer type that follows. +const SELF_AS_RAW = [ + String.raw`${PATH}from_mut${TURBOFISH}\(\s*self\s*\)`, + String.raw`self\s+as\s+\*mut\b[^,;)]*`, + String.raw`&raw\s+mut\s+\*\s*self\b(?!\s*\.)`, + String.raw`${PATH}addr_of_mut!\s*\(\s*\*\s*self\s*\)`, +].join("|"); + +// `self` as a `NonNull`: from the reference itself (`NonNull::from(self)`, +// `NonNull::from(&mut *self)`; the `\)` right after `self` keeps +// `NonNull::from(&mut *self.field)` out) or wrapped around a raw spelling. +const SELF_AS_NONNULL = [ + String.raw`${PATH}NonNull::from\(\s*(?:&\s*mut\s+\*\s*)?self\s*\)`, + String.raw`${PATH}NonNull::(?:new_unchecked|new)${TURBOFISH}\(\s*(?:${SELF_AS_RAW})\s*\)`, +].join("|"); + +const SELF_AS_POINTER = `(?:${SELF_AS_RAW}|${SELF_AS_NONNULL})`; + +// Method calls that keep the address: how a `NonNull::new(..)` is unwrapped +// and how a binding is adapted to the queue's element type. +const SAME_ADDRESS = String.raw`(?:\s*\.\s*(?:cast(?:_mut|_const)?${TURBOFISH}\(\)|as_ptr\(\)|unwrap\(\)|expect\([^)]*\)))*`; + +// The argument has to end there: `.push(NonNull::from(self).foo)` is not +// this shape, and neither is `.push(p.field)` for a binding `p` below. +const ARG_END = String.raw`\s*[,)]`; + +const DIRECT = new RegExp(`${HAND_OVER}${SELF_AS_POINTER}${SAME_ADDRESS}${ARG_END}`, "g"); + +// A local bound to the receiver's address, then handed over further down the +// same function (which ends at the next `fn` item; a closure inside it is the +// same function for this purpose). `let p: *mut Self = self;` is the coercion +// spelling; without the annotation `let p = self;` is just another reference. +const BINDING_HEAD = String.raw`\blet\s+(?:mut\s+)?(\w+)\s*`; +const SELF_POINTER_BINDINGS = [ + new RegExp(`${BINDING_HEAD}(?::[^=;]*)?=\\s*${SELF_AS_POINTER}${SAME_ADDRESS}\\s*;`, "g"), + new RegExp(`${BINDING_HEAD}:\\s*\\*(?:mut|const)\\b[^=;]*=\\s*self\\s*;`, "g"), +]; +const FN_ITEM = /^[ \t]*(?:pub(?:\([^)]*\))?\s+)?(?:(?:const|async|unsafe|extern\s+"[^"]*")\s+)*fn\s+\w+/m; + +// The binding passed as is, or wrapped into a `NonNull` at the call. +function handOverOfBinding(name: string): RegExp { + const wrapped = String.raw`${PATH}NonNull::(?:new_unchecked|new|from)${TURBOFISH}\(\s*${name}${SAME_ADDRESS}\s*\)`; + return new RegExp(`${HAND_OVER}(?:${wrapped}|\\b${name})${SAME_ADDRESS}${ARG_END}`); +} + +/** Byte offsets (into `stripped`) of every banned hand-over in one file. */ +function findHandOvers(stripped: string): number[] { + const hits: number[] = []; + for (const m of stripped.matchAll(DIRECT)) hits.push(m.index); + for (const pattern of SELF_POINTER_BINDINGS) { + for (const binding of stripped.matchAll(pattern)) { + const start = binding.index + binding[0].length; + const rest = stripped.slice(start); + const fnEnd = rest.search(FN_ITEM); + const body = fnEnd === -1 ? rest : rest.slice(0, fnEnd); + const call = body.search(handOverOfBinding(binding[1])); + if (call !== -1) hits.push(start + call); + } + } + return hits.sort((a, b) => a - b); +} + +function lineOf(text: string, offset: number): number { + return text.slice(0, offset).split("\n").length; +} + +// Documented, ratcheted exceptions: files allowed to keep exactly N of the +// shape. Lower an entry when you convert one; do not add entries. +const ALLOW: Record = { + // `FilePoll::deinit_possibly_defer(&mut self, ..)` puts its own slot back + // (`Store::put`, which recycles it at once, freeing it if it was a heap + // spill, when the poll was never registered; otherwise it is queued and + // freed after the event loop turn). Every `FilePoll::deinit*` entry point is + // `&mut self` and reached from many owners, so the conversion is a change of + // its own: #37803. Delete both entries when it lands. + "src/io/posix_event_loop.rs": 1, + "src/io/windows_event_loop.rs": 1, +}; + +const counts: Record = {}; +const offenders: string[] = []; +let scanned = 0; +for (const abs of rustSources) { + const source = path.relative(root, abs).replaceAll(path.sep, "/"); + // `src/cli` is a symlink into `src/runtime/cli`; count each file once under + // its canonical path. + if (path.relative(root, realpathSync(abs)).replaceAll(path.sep, "/") !== source) continue; + if (tracked !== null && !tracked.has(source)) continue; + scanned++; + const content = await file(abs).text(); + // Strip full-line comments so prose mentions (including the in-tree comments + // describing this hazard) don't count. `[ \t]*`, not `\s*`: `\s` crosses + // newlines and would swallow blank lines, shifting the reported line numbers. + const stripped = content.replace(/^[ \t]*\/\/.*$/gm, ""); + for (const offset of findHandOvers(stripped)) { + counts[source] = (counts[source] ?? 0) + 1; + if (counts[source] > (ALLOW[source] ?? 0)) { + offenders.push(`${source}:${lineOf(stripped, offset)}`); + } + } +} + +test("scans a non-empty set of tracked Rust sources", () => { + // Guards against the tracked/realpath filters above over-firing and leaving + // nothing to scan, which would make the ban below pass vacuously. + expect(scanned).toBeGreaterThan(0); +}); + +test("the patterns match the banned spellings and nothing else", () => { + const banned = [ + // `TranspilerJob::dispatch_to_main_thread(&mut self)` as it was. + "let job = NonNull::from(&mut *self);\nunsafe { (*transpiler_store).queue.push(job) };", + // `PatchTask::run_from_thread_pool_impl(&mut self)` as it was. + "unsafe {\n (*mgr)\n .patch_task_queue\n .push(core::ptr::NonNull::from(&mut *self));\n PackageManager::wake_raw(mgr);\n}", + // `TranspilerJob::run_from_js_thread(&mut self)` as it was. + "unsafe {\n (*vm)\n .transpiler_store\n .store\n .put(std::ptr::from_mut::(self))\n};", + "pool.put(self as *mut Self);", + "let slot = ptr::from_mut(self);\nunsafe { (*store).put(slot) };", + // The allowlisted `FilePoll::deinit_possibly_defer`, posix and windows spellings. + "let this = ptr::NonNull::from(self);\nvm.file_polls_mut().put(this, vm, was_ever_registered);", + "let this: ptr::NonNull = ptr::NonNull::from(&mut *self);\nvm.file_polls_mut().put(this, vm, was_ever_registered);", + // Other spellings of the same thing. + "queue.push(NonNull::from(self));", + "queue.push(core::ptr::NonNull::from(&mut *self));", + "queue.push(NonNull::new_unchecked(std::ptr::from_mut::(self)));", + "queue.push(NonNull::new(self as *mut Self).unwrap());", + "queue.push(core::ptr::NonNull::new_unchecked(&raw mut *self));", + "queue.push(NonNull::new_unchecked(ptr::addr_of_mut!(*self)).cast());", + "pending.push(ptr::from_mut(self));", + "pending.push(self as *mut Self as *mut c_void);", + "pending.push(\n NonNull::from(&mut *self),\n);", + // Through a local. + "let this = std::ptr::from_mut::(self);\nlet x = 1;\nqueue.push(NonNull::new_unchecked(this));", + "let this: *mut Self = self;\nif done {\n return;\n}\nqueue.push(NonNull::new(this).unwrap());", + "let this = NonNull::from(self).cast::();\nqueue.push(this.cast());", + "let mut this = NonNull::from(&mut *self);\n(*vm).store.queue.push(this, );", + ]; + const allowed = [ + // The raw-pointer shape the fix produces. + "unsafe { (*transpiler_store).queue.push(NonNull::new_unchecked(this)) };", + "(*mgr)\n .patch_task_queue\n .push(core::ptr::NonNull::new_unchecked(this));", + "unsafe { (*vm).transpiler_store.store.put(this) };", + "self.store.put(job);", + // A local reference parameter, moved into the push. + "installer.task_queue.push(core::ptr::NonNull::from(this));", + // `self` as a key or a context, not as storage. + "bun_core::handle_oom(handles.put(self, ()));", + 'err.put(self, b"name", name_value);', + "unsafe { self.owner.put(self.slot.as_ptr()) };", + // Something the receiver owns. + "queue.push(NonNull::from(&mut *self.inner));", + "queue.push(NonNull::from(&mut self.task));", + "batch.push(Batch::from(&raw mut self.task));", + "batch.push(Batch::from(core::ptr::addr_of_mut!(self.task)));", + "self.tasks.push(NonNull::new_unchecked(concurrent));", + "self.entries.push(item);", + "out.push(self.len);", + "let job = NonNull::from(&mut *self.job);\nqueue.push(job);", + // A helper-produced pointer and the shared spellings are out of scope. + "queue.push(self.as_non_null());", + "queue.push(NonNull::from(&*self));", + "let me = core::ptr::from_ref(self).cast_mut();\nENDPOINT_REGISTRY.with_borrow_mut(|v| {\n if !v.contains(&me) {\n v.push(me);\n }\n});", + // A self pointer that is not pushed, or whose name is only a prefix of + // what is pushed. + "let this = std::ptr::from_mut(self);\nSelf::finalize(this);", + "let this = std::ptr::from_mut(self);\nqueue.push(this_task);", + "let this = std::ptr::from_mut(self);\nqueue.push(NonNull::from(&mut (*this).task));", + // The binding's function ends before the push. + "let this = std::ptr::from_mut(self);\n }\n\n fn other(&mut self, this: NonNull) {\n self.queue.push(this);", + ]; + expect(banned.filter(s => findHandOvers(s).length === 0)).toEqual([]); + expect(allowed.filter(s => findHandOvers(s).length !== 0)).toEqual([]); +}); + +test("no method pushes or puts its own receiver", () => { + expect(offenders).toEqual([]); +}); + +test("allowlisted files still carry exactly their documented count", () => { + // Ratchet: once an allowlisted instance is converted, delete its entry so + // a new one cannot take its place. + const actual = Object.fromEntries(Object.keys(ALLOW).map(f => [f, counts[f] ?? 0])); + expect(actual).toEqual(ALLOW); +}); diff --git a/test/js/bun/resolve/concurrent-dynamic-import.test.ts b/test/js/bun/resolve/concurrent-dynamic-import.test.ts index b21ead8ddde6..b76ffa541458 100644 --- a/test/js/bun/resolve/concurrent-dynamic-import.test.ts +++ b/test/js/bun/resolve/concurrent-dynamic-import.test.ts @@ -1,12 +1,13 @@ import { expect, test } from "bun:test"; -import { bunEnv, bunExe, tempDir } from "harness"; +import { bunEnv, bunExe, isDebug, tempDir } from "harness"; +import { join } from "path"; // Two dynamic imports of the same specifier issued before the first async // transpile/fetch settles must both resolve. Under the new C++ module loader // each call gets its own embedder fetch promise (the registry entry is created // only after the first fetch settles), so the loser of that race must still be // resolved by Bun__onFulfillAsyncModule rather than left pending forever. -test("concurrent dynamic imports of the same module both resolve", async () => { +test.concurrent("concurrent dynamic imports of the same module both resolve", async () => { using dir = tempDir("concurrent-dyn-import", { "shared.ts": `export const heavy = "H";`, "modules.ts": `import { heavy } from "./shared";\nexport const lazy = heavy + "-lazy";`, @@ -30,3 +31,54 @@ test("concurrent dynamic imports of the same module both resolve", async () => { expect(stdout.trim()).toBe("ok"); expect(exitCode).toBe(0); }); + +// All of these import()s claim their transpiler-store job before the first one +// is handed back, so they overflow the store's inline hive: the spilled jobs +// are freed, not recycled, when the JS thread takes them back. The ones that +// fail to parse reach that hand-back through the transpiler's early return. +// Every import must settle with its own module's result, and (debug builds log +// each job the store takes on) every one of them must have gone through the +// store rather than the on-thread fallback, or the test exercises nothing. +test.concurrent("more concurrent dynamic imports than the transpiler store keeps inline all settle", async () => { + const hiveCap = 64; // TRANSPILER_JOB_HIVE_CAP in src/jsc/RuntimeTranspilerStore.rs + const count = hiveCap + 32; + const failsToParse = (i: number) => i % 8 === 7; + const files: Record = { + "entry.mjs": ` + const settled = await Promise.allSettled( + Array.from({ length: ${count} }, (_, i) => import("./mod" + i + ".ts")), + ); + console.log( + JSON.stringify( + settled.map(r => + r.status === "fulfilled" ? r.value.value : r.reason.name + ":" + r.reason.errors[0].constructor.name, + ), + ), + ); + `, + }; + for (let i = 0; i < count; i++) { + files[`mod${i}.ts`] = failsToParse(i) + ? `export const value: number = ${i}; export {` + : `export const value = ${i};`; + } + using dir = tempDir("many-concurrent-dyn-imports", files); + const debugLog = join(String(dir), "debug.log"); + + await using proc = Bun.spawn({ + cmd: [bunExe(), "entry.mjs"], + cwd: String(dir), + env: { ...bunEnv, ...(isDebug ? { BUN_DEBUG: debugLog, BUN_DEBUG_RuntimeTranspilerStore: "1" } : {}) }, + stdio: ["ignore", "pipe", "pipe"], + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(JSON.parse(stdout)).toEqual( + Array.from({ length: count }, (_, i) => (failsToParse(i) ? "AggregateError:BuildMessage" : i)), + ); + expect(stderr).toBe(""); + expect(exitCode).toBe(0); + if (isDebug) { + const log = await Bun.file(debugLog).text(); + expect(log.split("transpile(").length - 1).toBe(count); + } +}); diff --git a/test/js/web/workers/worker.test.ts b/test/js/web/workers/worker.test.ts index 49e7a197ed19..cddb8ee9f457 100644 --- a/test/js/web/workers/worker.test.ts +++ b/test/js/web/workers/worker.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import { once } from "events"; -import { bunEnv, bunExe, tempDir } from "harness"; +import { readFileSync } from "fs"; +import { bunEnv, bunExe, isASAN, isDebug, isLinux, tempDir } from "harness"; import path from "path"; import wt from "worker_threads"; @@ -536,6 +537,72 @@ describe("web worker", () => { expect(stdout).toBe("PASS\n"); expect(exitCode).toBe(0); }); + + // A worker's import()s are transpiled on the thread pool, in job slots that + // live inside the worker's VM, and come back to the JS thread in batches. + // Each worker keeps 64 imports in flight (every module that evaluates + // imports the next one; the query string makes each a fresh transpile) and + // asks to be terminated once the first one has evaluated, so the termination + // lands while the JS thread is part-way through a batch: the rest of that + // batch has to be released on the spot (the teardown never sees it), the + // jobs still queued or still out on the pool by the teardown. Under ASAN the + // host is leak-checked for all of them, minus the suppression covering + // everything allocated inside a transpile, which would hide the jobs' own + // storage; debug builds also log each job the store takes on, which rules + // out the imports having gone through the on-thread fallback instead. + const leakCheck = isASAN && isLinux; + test( + "terminate() while the worker's imports are still being transpiled", + async () => { + const files: Record = { + "worker.js": ` + let n = 0, posted = false; + function next() { const i = n++; import("./mod" + (i % 16) + ".js?v=" + i).then(done, done); } + function done() { if (!posted) { posted = true; postMessage("streaming"); } next(); } + for (let k = 0; k < 64; k++) next(); + `, + }; + for (let i = 0; i < 16; i++) files[`mod${i}.js`] = `export const value = ${i};`; + if (leakCheck) { + files["leaksan.supp"] = readFileSync(path.join(import.meta.dirname, "../../../leaksan.supp"), "utf8") + .split("\n") + .filter(line => line !== "leak:Bun__transpileFile") + .join("\n"); + } + using dir = tempDir("worker-transpile-churn", files); + const debugLog = path.join(String(dir), "debug.log"); + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `await Promise.all(Array.from({ length: 3 }, () => new Promise(res => { + const w = new Worker(process.argv[1]); w.addEventListener("close", res); + w.onmessage = () => w.terminate() }))); + console.log("PASS");`, + path.join(String(dir), "worker.js"), + ], + env: { + ...bunEnv, + ...(leakCheck + ? { + BUN_DESTRUCT_VM_ON_EXIT: "1", + ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "detect_leaks=1"].filter(Boolean).join(":"), + LSAN_OPTIONS: `print_suppressions=0:suppressions=${path.join(String(dir), "leaksan.supp")}`, + } + : {}), + ...(isDebug ? { BUN_DEBUG: debugLog, BUN_DEBUG_RuntimeTranspilerStore: "1" } : {}), + }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr, exitCode }).toEqual({ stdout: "PASS\n", stderr: "", exitCode: 0 }); + if (isDebug) expect(readFileSync(debugLog, "utf8")).toContain("transpile("); + }, + // A leak report's symbolization alone takes tens of seconds on the debug + // binary; the failure mode must get to print it. + leakCheck ? 90_000 : undefined, + ); }); });