Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -306,4 +306,6 @@

- Return `undefined` for `stat` paths through regular files with `throwIfNoEntry: false`, while preserving `lstat` errors. Ports [oven-sh/bun#39339](https://github.com/oven-sh/bun/pull/39339); thanks robobun.

- Close Worker-owned FileHandles and tracked fs descriptors on termination or natural exit, after pending filesystem work returns. Adapts [oven-sh/bun#34260](https://github.com/oven-sh/bun/pull/34260); thanks robobun.

- Settle child IPC send callbacks after peer disconnection, preserving immediate write errors and completing queued sends at channel close. Adapts [oven-sh/bun#43000](https://github.com/oven-sh/bun/pull/43000); thanks robobun.
2 changes: 2 additions & 0 deletions docs/runtime/nodejs-compat.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -345,6 +345,8 @@ Valid `vm.Script` cached data is accepted when JSC runs without a JIT. Validatio

### [`node:worker_threads`](https://nodejs.org/api/worker_threads.html)

Worker termination and natural exit close FileHandles opened by the Worker. Raw descriptors from `fs.open()` and `fs.openSync()` close by default; `trackUnmanagedFds: false` preserves them. Parent-owned descriptors remain open. FileHandle transfer through `postMessage()` remains unsupported.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Remove the stale compatibility-table claim

This newly added paragraph says trackUnmanagedFds is implemented, but the same node:worker_threads section still says at line 388 that Bun ignores the option. Users therefore receive mutually exclusive compatibility guidance on one page; update the existing limitation entry as part of this behavior change.

AGENTS.md reference: AGENTS.md:L187-L191

Useful? React with 👍 / 👎.


Worker shutdown releases the thread's native event-name cache after destroying its VM.

Worker exit removes `message` and `messageerrors` listeners before emitting `exit`, then releases user listeners after exit handlers return; unlike Node.js, Bun keeps its two internal `newListener` and `removeListener` hooks visible in `eventNames()` to avoid delayed owner-exit memory release on Windows ARM64.
Expand Down
5 changes: 4 additions & 1 deletion packages/bun-types/bun.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -701,9 +701,12 @@ declare module "bun" {
credentials?: import("undici-types").RequestCredentials | undefined;

/**
* Track file descriptors opened via `fs.open()` / `fs.openSync()` and
* close any that are still open when the Worker exits.
*
* @default true
*/
// trackUnmanagedFds?: boolean;
trackUnmanagedFds?: boolean | undefined;
// resourceLimits?: import("worker_threads").ResourceLimits;

/**
Expand Down
1 change: 1 addition & 0 deletions scripts/openclaw-ci/tests.ts
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ export const windowsSmoke = [
"test/js/node/worker_threads/worker-async-dispose.test.ts",
"test/js/node/worker_threads/worker-shutdown-post-leak.test.ts",
"test/js/node/worker_threads/worker-top-level-await.test.ts",
"test/js/node/worker_threads/worker-track-unmanaged-fds.test.ts",
"test/js/node/worker_threads/worker-transfer-terminate-stress.test.ts",
"test/js/node/worker_threads/worker_destruction.test.ts",
"test/js/node/worker_threads/worker_heap_snapshot_gc.test.ts",
Expand Down
2 changes: 2 additions & 0 deletions src/js/internal/fs/binding.ts
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,8 @@ interface FsBinding
realpathNativeSync: NodeFS["realpathSync"]["native"];
rmdir(path: import("node:fs").PathLike, options?: import("node:fs").RmOptions): Promise<void>;
rmdirSync(path: import("node:fs").PathLike, options?: import("node:fs").RmOptions): void;
trackFd(fd: number): void;
untrackFd(fd: number): void;
watch(path: string, options: import("node:fs").WatchOptions, listener: WatchListener): $ZigGeneratedClasses.FSWatcher;
watchFile(
path: string,
Expand Down
3 changes: 3 additions & 0 deletions src/js/node/fs.promises.ts
Original file line number Diff line number Diff line change
Expand Up @@ -434,6 +434,7 @@ function asyncWrap(fn: any, name: string) {
this[kClosePromise] = null;
this[kFlag] = flag;
if (this[kFd] !== -1) {
fs.trackFd(fd);
(fileHandleRegistry ??= new FinalizationRegistry(onFileHandleCollected)).register(this, { fd, path }, this);
}
}
Expand Down Expand Up @@ -1518,6 +1519,7 @@ function asyncWrap(fn: any, name: string) {
const flag = this[kFlag];
this[kFd] = -1;
fileHandleRegistry?.unregister(this);
fs.untrackFd(fd);
return {
data: { fd, flag },
deserializeInfo: "internal/fs/promises:FileHandle",
Expand All @@ -1532,6 +1534,7 @@ function asyncWrap(fn: any, name: string) {
this[kFd] = fd;
this[kFlag] = flag;
if (fd !== -1) {
fs.trackFd(fd);
(fileHandleRegistry ??= new FinalizationRegistry(onFileHandleCollected)).register(
this,
{ fd, path: undefined },
Expand Down
43 changes: 43 additions & 0 deletions src/jsc/VirtualMachine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -366,6 +366,10 @@ pub struct VirtualMachine {
/// The door out of this thread (`bun_jsc::vm_handle`): tickets for work
/// that leaves it are taken here, and `teardown` waits on it.
handle: core::mem::ManuallyDrop<crate::VmHandle>,
/// fds to close at worker exit (`Some` on worker threads only).
pub exit_fds: Option<Vec<bun_sys::Fd>>,
/// node:worker_threads `trackUnmanagedFds`.
pub track_unmanaged_fds: bool,
pub pending_ipc: Option<PendingIpc>,
pub hot_reload_counter: u32,

Expand Down Expand Up @@ -2536,6 +2540,13 @@ impl VirtualMachine {
rare.release_js_handles();
}
}
// Off-thread fs work on these fds is back and script is forbidden.
// SAFETY: fn contract (statement-scoped exclusive access).
if let Some(fds) = unsafe { (*this).exit_fds.take() } {
for fd in fds {
let _ = bun_sys::FdExt::close_allowing_standard_io(fd, None);
}
}
teardown_log!("teardown: script forbidden, resources cancelled, off-thread work back");

// ---- C. JSC VM -------------------------------------------------------
Expand Down Expand Up @@ -3323,6 +3334,8 @@ impl VirtualMachine {
addr_of_mut!((*vm).argv).write(Vec::new());
addr_of_mut!((*vm).argv1).write(None);
addr_of_mut!((*vm).resolved_path_dups).write(Vec::new());
addr_of_mut!((*vm).exit_fds).write(None);
addr_of_mut!((*vm).track_unmanaged_fds).write(false);
addr_of_mut!((*vm).macros).write(Default::default());
addr_of_mut!((*vm).macro_entry_points).write(Default::default());
addr_of_mut!((*vm).auto_killer).write(Default::default());
Expand Down Expand Up @@ -6223,6 +6236,36 @@ impl VirtualMachine {
Ok(self.pending_internal_promise().unwrap())
}

/// A raw `fs.open` fd reached user code.
#[inline]
pub fn track_unmanaged_fd(&mut self, fd: bun_sys::Fd) {
if self.track_unmanaged_fds {
self.track_managed_fd(fd);
}
}

/// A FileHandle owns `fd`; Node closes those at worker exit regardless of the option.
#[inline]
pub fn track_managed_fd(&mut self, fd: bun_sys::Fd) {
if let Some(set) = self.exit_fds.as_mut() {
if !set.contains(&fd) {
set.push(fd);
}
}
}

/// `fd` was closed, or transferred to another thread.
#[inline]
pub fn untrack_fd(&mut self, fd: bun_sys::Fd) -> bool {
if let Some(set) = self.exit_fds.as_mut() {
if let Some(i) = set.iter().position(|&f| f == fd) {
set.swap_remove(i);
return true;
}
}
false
}

/// Tracks a listening socket so watch-mode reloads can close it.
pub fn add_listening_socket_for_watch_mode(&mut self, socket: bun_sys::Fd) {
if self.hot_reload != HotReload::Watch && !self.test_isolation_enabled {
Expand Down
6 changes: 6 additions & 0 deletions src/jsc/bindings/webcore/JSWorker.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -402,6 +402,12 @@ template<> __attribute__((minsize)) JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES
}
}

auto trackUnmanagedFds = optionsObject->getIfPropertyExists(lexicalGlobalObject, Identifier::fromString(vm, "trackUnmanagedFds"_s));
RETURN_IF_EXCEPTION(throwScope, {});
if (trackUnmanagedFds && !trackUnmanagedFds.isUndefinedOrNull()) {
options.trackUnmanagedFds = trackUnmanagedFds.toBoolean(lexicalGlobalObject);
}

auto preloadModulesValue = optionsObject->getIfPropertyExists(lexicalGlobalObject, Identifier::fromString(vm, "preload"_s));
RETURN_IF_EXCEPTION(throwScope, {});
if (preloadModulesValue) {
Expand Down
2 changes: 2 additions & 0 deletions src/jsc/bindings/webcore/WorkerMessagingProxy.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ void* WebWorker__create(
bool unrefByDefault,
bool evalMode,
bool isNodeWorker,
bool trackUnmanagedFds,
StringImpl** argvPtr,
size_t argvLen,
bool defaultExecArgv,
Expand Down Expand Up @@ -274,6 +275,7 @@ ExceptionOr<void> WorkerMessagingProxy::startWorkerGlobalScope(const String& scr
m_options.unref,
m_options.evalMode,
m_options.kind == WorkerOptions::Kind::Node,
m_options.trackUnmanagedFds,
reinterpret_cast<WTF::StringImpl**>(m_options.argv.begin()),
m_options.argv.size(),
m_options.inheritExecArgv,
Expand Down
2 changes: 2 additions & 0 deletions src/jsc/bindings/webcore/WorkerOptions.h
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ struct WorkerOptions {
// true, then we need to make sure that `process.argv` contains "[worker eval]" instead of the
// Blob URL.
bool evalMode { false };
// node:worker_threads `trackUnmanagedFds`; default matches Node.js.
bool trackUnmanagedFds { true };
Kind kind { Kind::Web };
// Serialized array containing [workerData, environmentData]
// (environmentData is always a Map)
Expand Down
8 changes: 8 additions & 0 deletions src/jsc/web_worker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,9 @@ pub struct WebWorker {
/// Created by `node:worker_threads`' `Worker` (as opposed to the Web `Worker`
/// constructor): loads `node:worker_threads` before preloads and the entry point.
is_node_worker: bool,
/// node:worker_threads `trackUnmanagedFds`.
track_unmanaged_fds: bool,

/// Borrowed from the proxy's `WorkerOptions` (alive as long as the proxy).
argv_ptr: *const WTFStringImpl,
argv_len: usize,
Expand Down Expand Up @@ -362,6 +365,7 @@ impl WebWorker {
default_unref: bool,
eval_mode: bool,
is_node_worker: bool,
track_unmanaged_fds: bool,
argv_ptr: *const WTFStringImpl,
argv_len: usize,
inherit_exec_argv: bool,
Expand Down Expand Up @@ -591,6 +595,8 @@ impl WebWorker {
mini,
eval_mode,
is_node_worker,
track_unmanaged_fds: track_unmanaged_fds || parent_ref.track_unmanaged_fds,

argv_ptr,
argv_len,
exec_argv_ptr,
Expand Down Expand Up @@ -985,6 +991,8 @@ impl WebWorker {
vm_ref.cpu_profiler_config = Some(config);
crate::bun_cpu_profiler::start_cpu_profiler(vm_ref.jsc_vm_mut());
}
vm_ref.exit_fds = Some(Vec::new());
vm_ref.track_unmanaged_fds = self.track_unmanaged_fds;
}

// Publish now (rather than at the end of startVM) so that:
Expand Down
2 changes: 2 additions & 0 deletions src/runtime/node/node.classes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,8 @@ export default [
symlinkSync: { async: false, fn: "symlinkSync", length: 3 },
truncate: { async: true, fn: "truncate", length: 3 },
truncateSync: { async: false, fn: "truncateSync", length: 2 },
trackFd: { async: false, fn: "trackFd", length: 1 },
untrackFd: { async: false, fn: "untrackFd", length: 1 },
unlink: { async: true, fn: "unlink", length: 2 },
unlinkSync: { async: false, fn: "unlinkSync", length: 1 },
utimes: { async: true, fn: "utimes", length: 4 },
Expand Down
58 changes: 47 additions & 11 deletions src/runtime/node/node_fs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -638,8 +638,9 @@ mod _async_tasks {
pub(crate) type UVFSRequest<R, A, const F: NodeFSFunctionEnum> = AsyncFSTask<R, A, F>;

#[cfg(windows)]
pub(crate) struct UVFSRequest<R, A, const F: NodeFSFunctionEnum> {
pub(crate) struct UVFSRequest<R: FsReturn, A: FsArgument, const F: NodeFSFunctionEnum> {
pub(crate) completion: FsCompletion,

pub args: ThreadIsolated<A>,
pub(crate) global_object: bun_ptr::BackRef<JSGlobalObject>,
pub(crate) req: uv::fs_t,
Expand Down Expand Up @@ -979,6 +980,11 @@ mod _async_tasks {
// SAFETY: caller guarantees `this` is the live Box-leaked allocation;
// reclaim ownership (paired with the Box::leak in create()).
let mut task = unsafe { bun_core::heap::take(this) };
if task.req.result.int() == i64::from(uv::UV_ECANCELED) {
if let Some(fd) = task.args.closed_fd() {
VirtualMachine::get().as_mut().track_managed_fd(fd);
}
}
// A result nobody took (the request's context stopped: released unrun).
if let Ok(result) = core::mem::replace(&mut task.result, Err(sys::Error::default())) {
result.discard();
Expand Down Expand Up @@ -1014,6 +1020,22 @@ mod _async_tasks {
fn signal(&self) -> Option<&AbortSignal> {
None
}
/// The fd this call closes; re-tracked if the job is handed back unrun.
fn closed_fd(&self) -> Option<FD> {
None
}
}

// SAFETY: `Close` is a bare fd.
unsafe impl ThreadIsolatedArg for args::Close {}
impl FsArgument for args::Close {
#[inline]
fn from_js(ctx: &JSGlobalObject, arguments: &mut ArgumentsSlice) -> JsResult<Self> {
args::Close::from_js(ctx, arguments)
}
fn closed_fd(&self) -> Option<FD> {
self.was_tracked.then_some(self.fd)
}
}

/// Forward [`FsArgument`] to the inherent `from_js` each `args::*` struct
Expand Down Expand Up @@ -1064,7 +1086,6 @@ mod _async_tasks {
args::Futimes,
args::FdataSync,
args::Fsync,
args::Close,
);
// `ReadFile`/`WriteFile` carry an `AbortSignal` field — opt them in so the
// `const _ = assert!(…::HAVE_ABORT_SIGNAL)` invariants in `async_` hold and
Expand Down Expand Up @@ -1150,7 +1171,12 @@ mod _async_tasks {
impl FsReturn for FD {
#[inline]
fn fs_to_js(self, global: &JSGlobalObject) -> JsResult<JSValue> {
Ok(crate::node::types::FdJsc::to_js(self, global))
let value = crate::node::types::FdJsc::to_js(self, global);
// Windows converts native handles to CRT fds here; close parses that JS-visible fd.
if let Some(fd) = crate::node::types::FdJsc::from_js(value) {
global.bun_vm().as_mut().track_unmanaged_fd(fd);
}
Ok(value)
}
#[inline]
fn discard(self) {
Expand Down Expand Up @@ -1224,24 +1250,29 @@ mod _async_tasks {

/// One `fs.promises.*` operation on the work pool. The arguments' JS-backed
/// buffers are pinned and rooted (`ThreadIsolated`) and read under the job's ticket.
pub(crate) struct AsyncFSTask<R: FsReturn, A, const F: NodeFSFunctionEnum> {
pub(crate) struct AsyncFSTask<R: FsReturn, A: FsArgument, const F: NodeFSFunctionEnum> {
pub args: ThreadIsolated<A>,
pub(crate) result: Maybe<R>,
/// The pool half executed (its completion may still be released unrun).
ran: bool,
}
impl<R: FsReturn, A, const F: NodeFSFunctionEnum> Drop for AsyncFSTask<R, A, F> {
/// A job whose context stopped is dropped with its result untaken.

impl<R: FsReturn, A: FsArgument, const F: NodeFSFunctionEnum> Drop for AsyncFSTask<R, A, F> {
fn drop(&mut self) {
if !self.ran {
if let Some(fd) = self.args.closed_fd() {
VirtualMachine::get().as_mut().track_managed_fd(fd);
}
}
// A job whose context stopped is dropped with its result untaken.
if let Ok(result) = core::mem::replace(&mut self.result, Err(sys::Error::default())) {
result.discard();
}
}
}
// SAFETY: results are plain data / owned buffers / WTF strings built off
// thread for hand-off (`ret::*`); `ThreadIsolated<A>` is Send by its contract.
unsafe impl<R: FsReturn, A: ThreadIsolatedArg, const F: NodeFSFunctionEnum> Send
for AsyncFSTask<R, A, F>
{
}
unsafe impl<R: FsReturn, A: FsArgument, const F: NodeFSFunctionEnum> Send for AsyncFSTask<R, A, F> {}

/// How an async fs operation reports its result: a promise, or the `node:fs` callback, called from the completion.
#[derive(bun_jsc::JsAffine)]
Expand Down Expand Up @@ -1372,6 +1403,7 @@ mod _async_tasks {
) -> Option<bun_jsc::Completion<Self>> {
let mut node_fs = NodeFS::default();
this.result = NodeFS::dispatch::<R, A, F>(&mut node_fs, &this.args, Flavor::Async);
this.ran = true;
Some(done)
}

Expand Down Expand Up @@ -1440,6 +1472,7 @@ mod _async_tasks {
// Sentinel — overwritten by `run` before any read. `Maybe<R>`
// may be niche-optimised; never construct an all-zero `Result`.
result: Err(sys::Error::default()),
ran: false,
},
AsyncFSJs {
completion,
Expand Down Expand Up @@ -3603,14 +3636,17 @@ pub(crate) mod args {

pub(crate) struct Close {
pub(crate) fd: FD,
pub(crate) was_tracked: bool,
}
impl Close {
pub(crate) fn from_js(
ctx: &JSGlobalObject,
arguments: &mut ArgumentsSlice,
) -> JsResult<Close> {
let fd = FD::from_js_required(ctx, arguments)?;
Ok(Close { fd })
// Before the close runs, as Node does, so a reuse of the number is tracked fresh.
let was_tracked = ctx.bun_vm().as_mut().untrack_fd(fd);
Ok(Close { fd, was_tracked })
}
}

Expand Down
24 changes: 24 additions & 0 deletions src/runtime/node/node_fs_binding.rs
Original file line number Diff line number Diff line change
Expand Up @@ -348,6 +348,30 @@ impl Binding {
Ok(res) => Ok(res),
}
}

/// A FileHandle took ownership of the fd (constructor, `kDeserialize`).
pub(crate) fn track_fd(
_this: &Self,
global: &JSGlobalObject,
frame: &CallFrame,
) -> JsResult<JSValue> {
if let Some(fd) = <bun_sys::Fd as bun_sys_jsc::FdJsc>::from_js(frame.argument(0)) {
global.bun_vm().as_mut().track_managed_fd(fd);
}
Ok(JSValue::UNDEFINED)
}

/// A FileHandle gave up its fd without closing it (`kTransfer`).
pub(crate) fn untrack_fd(
_this: &Self,
global: &JSGlobalObject,
frame: &CallFrame,
) -> JsResult<JSValue> {
if let Some(fd) = <bun_sys::Fd as bun_sys_jsc::FdJsc>::from_js(frame.argument(0)) {
global.bun_vm().as_mut().untrack_fd(fd);
}
Ok(JSValue::UNDEFINED)
}
}

/// Generates the `pub const <name> = call{Async,Sync}(.<fn>)` block.
Expand Down
Loading
Loading