diff --git a/src/js/internal/fs/binding.ts b/src/js/internal/fs/binding.ts index f8cccca13bbc..53f28693a104 100644 --- a/src/js/internal/fs/binding.ts +++ b/src/js/internal/fs/binding.ts @@ -123,4 +123,12 @@ interface FsBinding writev(fd: number, buffers: readonly NodeJS.ArrayBufferView[], position?: number | null): Promise; } -export default $rust("node_fs_binding.rs", "createBinding") as FsBinding; +// `xCb(callback, ...args)`: the promise arm's arguments after the callback. `fs.cp` is a JS chain, so no `cpCb`. +type FsCallbackArm = { + [K in Exclude as T[K] extends (...args: any[]) => Promise ? `${K}Cb` : never]: ( + callback: (err: any, value?: any) => void, + ...args: any[] + ) => void; +}; + +export default $rust("node_fs_binding.rs", "createBinding") as FsBinding & FsCallbackArm; diff --git a/src/js/internal/trace_events.ts b/src/js/internal/trace_events.ts index 8d040b58e3cc..13301c34ca9e 100644 --- a/src/js/internal/trace_events.ts +++ b/src/js/internal/trace_events.ts @@ -626,6 +626,8 @@ function installFsInstrumentation() { for (const method in asyncOps) { const original = binding[method]; if (typeof original === "function") binding[method] = wrapFsAsyncMethod(original, asyncOps[method]); + const originalCb = binding[method + "Cb"]; + if (typeof originalCb === "function") binding[method + "Cb"] = wrapFsCallbackMethod(originalCb, asyncOps[method]); } // fs.opendir never reaches the binding (Bun's Dir reads lazily), so wrap the // node:fs export for the node.fs_dir.async category. @@ -690,6 +692,27 @@ function wrapFsAsyncMethod(original, names: string[]) { }; } +// The callback arm of the binding (`xCb(callback, ...args)`): the span ends when the operation calls back. +function wrapFsCallbackMethod(original, names: string[]) { + return function (callback, ...args) { + if (suppressFsEvents || !isCategoryGroupEnabled(kFsAsyncCat)) return original.$apply(this, arguments); + for (let i = 0; i < names.length; i++) emitEvent("b", kFsAsyncCat, names[i]); + try { + return original.$call( + this, + function () { + emitFsAsyncEnd(names); + return callback.$apply(undefined, arguments); + }, + ...args, + ); + } catch (err) { + emitFsAsyncEnd(names); + throw err; + } + }; +} + // The global console is native (not the JS Console class), so wrap its // counter/timer methods directly. Counts and timer labels are tracked in // parallel maps because the native implementations don't expose theirs; diff --git a/src/js/node/fs.ts b/src/js/node/fs.ts index 07e4f126b202..8503df3b2d45 100644 --- a/src/js/node/fs.ts +++ b/src/js/node/fs.ts @@ -24,8 +24,11 @@ function lazyGlob() { const { guardCallback, kCustomPromisifyArgsSymbol } = require("internal/shared"); +// Captured at load: a process.nextTick that user code replaces later (fake timers) must not hold back fs callbacks. +const nextTick = process.nextTick; + // guardCallback reroutes a throw inside the user callback to the -// uncaughtException path instead of rejecting the internal promise chain. +// uncaughtException path. function wrapFsCallback(callback) { return guardCallback(callback); } @@ -40,16 +43,19 @@ function ensureCallback(callback) { return wrapFsCallback(callback); } -// Micro-optimization: avoid creating a new function for every call -// bind() is slightly more optimized in JSC -// This code is equivalent to: -// -// function () { callback(null); } -// -function nullcallback(callback) { - return FunctionPrototypeBind.$call(callback, undefined, null); +// Via nextTick: called from the reaction itself (a microtask), a callback's microtasks would run before its nextTicks. +function settleCallback(promise, callback) { + promise.then( + value => nextTick(callback, null, value), + err => nextTick(callback, err), + ); +} +function settleCallbackWithNull(promise, callback) { + promise.then( + () => nextTick(callback, null), + err => nextTick(callback, err), + ); } -const FunctionPrototypeBind = nullcallback.bind; function openAsBlob(path, options) { return Promise.$resolve(Bun.file(path, options)); @@ -68,7 +74,7 @@ var access = function access(path, mode, callback?) { } callback = ensureCallback(callback); - fs.access(path, mode).then(callback, callback); + fs.accessCb(err => callback(err), path, mode); }, appendFile = function appendFile(path, data, options, callback?) { if (!$isCallable(callback)) { @@ -78,12 +84,12 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.appendFile(path, data, options).then(nullcallback(callback), callback); + fs.appendFileCb(callback, path, data, options); }, close = function close(fd, callback?) { if ($isCallable(callback)) { callback = wrapFsCallback(callback); - fs.close(fd).then(() => callback(null), callback); + fs.closeCb(err => callback(err), fd); } else if (callback === undefined) { fs.close(fd).then(() => {}); } else { @@ -98,7 +104,7 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); // route through promises.rm for the JS-side ERR_FS_EISDIR validation - require("node:fs/promises").rm(path, options).then(nullcallback(callback), callback); + settleCallback(require("node:fs/promises").rm(path, options), callback); }, rmdir = function rmdir(path, options, callback?) { if ($isCallable(options)) { @@ -108,10 +114,8 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); // Node 26 removed `recursive` (DEP0147), but packages still pass it. Keep it working through `rm`. - (options?.recursive ? require("node:fs/promises").rm(path, options) : fs.rmdir(path, options)).then( - nullcallback(callback), - callback, - ); + if (options?.recursive) settleCallback(require("node:fs/promises").rm(path, options), callback); + else fs.rmdirCb(callback, path, options); }, copyFile = function copyFile(src, dest, mode, callback?) { if ($isCallable(mode)) { @@ -121,16 +125,13 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.copyFile(src, dest, mode).then(nullcallback(callback), callback); + fs.copyFileCb(callback, src, dest, mode); }, exists = function exists(path, callback) { callback = ensureCallback(callback); try { - fs.exists.$apply(fs, [path]).then( - existed => callback(existed), - _ => callback(false), - ); + fs.existsCb((err, existed) => callback(err ? false : existed), path); } catch { callback(false); } @@ -138,22 +139,22 @@ var access = function access(path, mode, callback?) { chown = function chown(path, uid, gid, callback) { callback = ensureCallback(callback); - fs.chown(path, uid, gid).then(nullcallback(callback), callback); + fs.chownCb(callback, path, uid, gid); }, chmod = function chmod(path, mode, callback) { callback = ensureCallback(callback); - fs.chmod(path, mode).then(nullcallback(callback), callback); + fs.chmodCb(callback, path, mode); }, fchmod = function fchmod(fd, mode, callback) { callback = ensureCallback(callback); - fs.fchmod(fd, mode).then(nullcallback(callback), callback); + fs.fchmodCb(callback, fd, mode); }, fchown = function fchown(fd, uid, gid, callback) { callback = ensureCallback(callback); - fs.fchown(fd, uid, gid).then(nullcallback(callback), callback); + fs.fchownCb(callback, fd, uid, gid); }, fstat = function fstat(fd, options, callback?) { if ($isCallable(options)) { @@ -162,14 +163,12 @@ var access = function access(path, mode, callback?) { } callback = wrapFsCallback(callback); - fs.fstat(fd, options).then(function (stats) { - callback(null, stats); - }, callback); + fs.fstatCb(callback, fd, options); }, fsync = function fsync(fd, callback) { callback = ensureCallback(callback); - fs.fsync(fd).then(nullcallback(callback), callback); + fs.fsyncCb(callback, fd); }, ftruncate = function ftruncate(fd, len = 0, callback?) { if ($isCallable(len)) { @@ -179,30 +178,30 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.ftruncate(fd, len).then(nullcallback(callback), callback); + fs.ftruncateCb(callback, fd, len); }, futimes = function futimes(fd, atime, mtime, callback) { callback = ensureCallback(callback); - fs.futimes(fd, atime, mtime).then(nullcallback(callback), callback); + fs.futimesCb(callback, fd, atime, mtime); }, lchmod = constants.O_SYMLINK !== undefined ? function lchmod(path, mode, callback) { callback = ensureCallback(callback); - fs.lchmod(path, mode).then(nullcallback(callback), callback); + fs.lchmodCb(callback, path, mode); } : undefined, // lchmod is only available on macOS lchown = function lchown(path, uid, gid, callback) { callback = ensureCallback(callback); - fs.lchown(path, uid, gid).then(nullcallback(callback), callback); + fs.lchownCb(callback, path, uid, gid); }, link = function link(existingPath, newPath, callback) { callback = ensureCallback(callback); - fs.link(existingPath, newPath).then(nullcallback(callback), callback); + fs.linkCb(callback, existingPath, newPath); }, mkdir = function mkdir(path, options, callback?) { if ($isCallable(options)) { @@ -212,7 +211,7 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.mkdir(path, options).then(nullcallback(callback), callback); + fs.mkdirCb(callback, path, options); }, mkdtemp = function mkdtemp(prefix, options, callback?) { if ($isCallable(options)) { @@ -222,9 +221,7 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.mkdtemp(prefix, options).then(function (folder) { - callback(null, folder); - }, callback); + fs.mkdtempCb(callback, prefix, options); }, open = function open(path, flags, mode?, callback?) { if (arguments.length < 3) { @@ -238,14 +235,12 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.open(path, flags, mode).then(function (fd) { - callback(null, fd); - }, callback); + fs.openCb(callback, path, flags, mode); }, fdatasync = function fdatasync(fd, callback) { callback = ensureCallback(callback); - fs.fdatasync(fd).then(nullcallback(callback), callback); + fs.fdatasyncCb(callback, fd); }, read = function read(fd, buffer, offsetOrOptions?, length?, position?, callback?) { // fd = getValidatedFd(fd); DEFERRED TO NATIVE @@ -287,14 +282,19 @@ var access = function access(path, mode, callback?) { throw $ERR_INVALID_ARG_TYPE("callback", "function", callback); } callback = wrapFsCallback(callback); - fs.read(fd, buffer, offset, length, position).then( - bytesRead => void callback(null, bytesRead, buffer), - err => callback(err), + fs.readCb( + (err, bytesRead) => (err ? callback(err) : callback(null, bytesRead, buffer)), + fd, + buffer, + offset, + length, + position, ); }, write = function write(fd, buffer, offsetOrOptions, length?, position?, callback?) { - function wrapper(bytesWritten) { - callback(null, bytesWritten, buffer); + function onWritten(err, bytesWritten) { + if (err) callback(err); + else callback(null, bytesWritten, buffer); } // $isTypedArrayView excludes DataView, so a DataView would fall through @@ -311,7 +311,7 @@ var access = function access(path, mode, callback?) { } = offsetOrOptions ?? {}); } - fs.write(fd, buffer, offsetOrOptions, length, position).then(wrapper, callback); + fs.writeCb(onWritten, fd, buffer, offsetOrOptions, length, position); return; } @@ -334,7 +334,7 @@ var access = function access(path, mode, callback?) { callback = position; callback = ensureCallback(callback); - fs.write(fd, buffer, offsetOrOptions, length).then(wrapper, callback); + fs.writeCb(onWritten, fd, buffer, offsetOrOptions, length); }, readdir = function readdir(path, options, callback?) { if ($isCallable(options)) { @@ -344,23 +344,19 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.readdir(path, options).then(function (files) { - callback(null, files); - }, callback); + fs.readdirCb(callback, path, options); }, readFile = function readFile(path, options, callback?) { callback ||= options; callback = ensureCallback(callback); - fs.readFile(path, options).then(function (data) { - callback(null, data); - }, callback); + fs.readFileCb(callback, path, options); }, writeFile = function writeFile(path, data, options, callback?) { callback ||= options; callback = ensureCallback(callback); - fs.writeFile(path, data, options).then(nullcallback(callback), callback); + fs.writeFileCb(callback, path, data, options); }, readlink = function readlink(path, options, callback?) { if ($isCallable(options)) { @@ -370,14 +366,12 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.readlink(path, options).then(function (linkString) { - callback(null, linkString); - }, callback); + fs.readlinkCb(callback, path, options); }, rename = function rename(oldPath, newPath, callback) { callback = ensureCallback(callback); - fs.rename(oldPath, newPath).then(nullcallback(callback), callback); + fs.renameCb(callback, oldPath, newPath); }, lstat = function lstat(path, options, callback?) { if ($isCallable(options)) { @@ -387,9 +381,7 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.lstat(path, options).then(function (stats) { - callback(null, stats); - }, callback); + fs.lstatCb(callback, path, options); }, stat = function stat(path, options, callback?) { if ($isCallable(options)) { @@ -401,13 +393,11 @@ var access = function access(path, mode, callback?) { const signal = options?.signal; if (signal?.aborted) { - process.nextTick(callback, $makeAbortError(undefined, { cause: signal.reason })); + nextTick(callback, $makeAbortError(undefined, { cause: signal.reason })); return; } - fs.stat(path, options).then(function (stats) { - callback(null, stats); - }, callback); + fs.statCb(callback, path, options); }, statfs = function statfs(path, options, callback?) { if ($isCallable(options)) { @@ -417,9 +407,7 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.statfs(path, options).then(function (stats) { - callback(null, stats); - }, callback); + fs.statfsCb(callback, path, options); }, symlink = function symlink(target, path, type, callback?) { if (callback === undefined) { @@ -431,7 +419,8 @@ var access = function access(path, mode, callback?) { callback = wrapFsCallback(callback); } - fs.symlink(target, path, type).then(callback, callback); + if ($isCallable(callback)) fs.symlinkCb(err => callback(err === null ? undefined : err), target, path, type); + else fs.symlink(target, path, type).then(callback, callback); }, truncate = function truncate(path, len, callback?) { if (typeof path === "number") { @@ -448,22 +437,22 @@ var access = function access(path, mode, callback?) { } callback = ensureCallback(callback); - fs.truncate(path, len).then(nullcallback(callback), callback); + fs.truncateCb(callback, path, len); }, unlink = function unlink(path, callback) { callback = ensureCallback(callback); - fs.unlink(path).then(nullcallback(callback), callback); + fs.unlinkCb(callback, path); }, utimes = function utimes(path, atime, mtime, callback) { callback = ensureCallback(callback); - fs.utimes(path, atime, mtime).then(nullcallback(callback), callback); + fs.utimesCb(callback, path, atime, mtime); }, lutimes = function lutimes(path, atime, mtime, callback) { callback = ensureCallback(callback); - fs.lutimes(path, atime, mtime).then(nullcallback(callback), callback); + fs.lutimesCb(callback, path, atime, mtime); }, accessSync = fs.accessSync.bind(fs), appendFileSync = fs.appendFileSync.bind(fs), @@ -599,7 +588,12 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.writev(fd, buffers, position).$then(bytesWritten => callback(null, bytesWritten, buffers), callback); + fs.writevCb( + (err, bytesWritten) => (err ? callback(err) : callback(null, bytesWritten, buffers)), + fd, + buffers, + position, + ); }, writevSync = fs.writevSync.bind(fs), readv = function readv(fd, buffers, position, callback?) { @@ -610,7 +604,7 @@ var access = function access(path, mode, callback?) { callback = ensureCallback(callback); - fs.readv(fd, buffers, position).$then(bytesRead => callback(null, bytesRead, buffers), callback); + fs.readvCb((err, bytesRead) => (err ? callback(err) : callback(null, bytesRead, buffers)), fd, buffers, position); }, readvSync = fs.readvSync.bind(fs), Dirent = fs.Dirent, @@ -813,9 +807,7 @@ const realpath: typeof import("node:fs").realpath = } callback = ensureCallback(callback); - fs.realpath(p, options).then(function (resolvedPath) { - callback(null, resolvedPath); - }, callback); + fs.realpathCb(callback, p, options); } as typeof import("node:fs").realpath) : (function realpath(p, options, callback) { if ($isCallable(options)) { @@ -873,7 +865,7 @@ const realpath: typeof import("node:fs").realpath = LOOP(); }); } else { - process.nextTick(LOOP); + nextTick(LOOP); } // Walk down the path, swapping out linked path parts for their real @@ -917,7 +909,7 @@ const realpath: typeof import("node:fs").realpath = // If not a symlink, skip to the next path part if (!stats.isSymbolicLink()) { knownHard.add(base); - return process.nextTick(LOOP); + return nextTick(LOOP); } // Stat & read the link if not read before. @@ -952,7 +944,7 @@ const realpath: typeof import("node:fs").realpath = LOOP(); }); } else { - process.nextTick(LOOP); + nextTick(LOOP); } } } as typeof import("node:fs").realpath); @@ -964,9 +956,7 @@ realpath.native = function realpath(p, options, callback?) { callback = ensureCallback(callback); - fs.realpathNative(p, options).then(function (resolvedPath) { - callback(null, resolvedPath); - }, callback); + fs.realpathNativeCb(callback, p, options); }; realpathSync.native = fs.realpathNativeSync.bind(fs); @@ -1008,7 +998,7 @@ function cp(src, dest, options, callback) { dest = getValidatedFsPath(dest, "dest"); callback = guardCallback(callback); - require("node:fs/promises").cp(src, dest, options).then(callOnceWithNull.bind(null, callback), callback); + settleCallbackWithNull(require("node:fs/promises").cp(src, dest, options), callback); } function _toUnixTimestamp(time: any, name = "time") { @@ -1032,19 +1022,13 @@ function _toUnixTimestamp(time: any, name = "time") { function onOpendirStatFulfilled(callback, path, result, stats) { if (!stats.isDirectory()) { - process.nextTick(callback, opendirNotDirError(path)); + nextTick(callback, opendirNotDirError(path)); return; } - process.nextTick(callback, null, result); + nextTick(callback, null, result); } function onOpendirStatRejected(callback, path, err) { - process.nextTick(callback, typeof err?.errno === "number" ? opendirStatError(err, path) : err); -} -function callOnceWithNull(callback) { - callback(null); -} -function callOnceWithNullThen(callback, value) { - callback(null, value); + nextTick(callback, typeof err?.errno === "number" ? opendirStatError(err, path) : err); } function opendirSync(path, options) { @@ -1169,7 +1153,7 @@ class Dir { validateFunction(cb, "callback"); cb = guardCallback(cb); // node's callback overload returns undefined (like close(cb) above) - this.read().then(callOnceWithNullThen.bind(null, cb), cb); + settleCallback(this.read(), cb); return; } @@ -1210,7 +1194,7 @@ class Dir { if (!$isUndefinedOrNull(cb)) { validateFunction(cb, "callback"); cb = guardCallback(cb); - this.close().then(callOnceWithNull.bind(null, cb), cb); + settleCallbackWithNull(this.close(), cb); return; } return this.#enqueue(this.#closeOp.bind(this)); @@ -1267,16 +1251,7 @@ function glob(pattern: string | string[], options, callback) { // the callback surfaces as an uncaught exception instead of rejecting the // internal promise chain (and is never routed back into `callback` as an // error), matching Node.js. - Array.fromAsync(lazyGlob().glob(pattern, options ?? kEmptyObject)).then( - nextTickWithNullThen.bind(null, callback), - nextTickWith.bind(null, callback), - ); -} -function nextTickWithNullThen(callback, result) { - process.nextTick(callback, null, result); -} -function nextTickWith(callback, err) { - process.nextTick(callback, err); + settleCallback(Array.fromAsync(lazyGlob().glob(pattern, options ?? kEmptyObject)), callback); } function globSync(pattern: string | string[], options): string[] { diff --git a/src/runtime/dispatch.rs b/src/runtime/dispatch.rs index 300784b6cbde..2f878fc4f13b 100644 --- a/src/runtime/dispatch.rs +++ b/src/runtime/dispatch.rs @@ -548,7 +548,8 @@ pub(crate) fn run_task( for_each_fs_uv_op!(__fs_pat) => { macro_rules! __fs_run { ($($tag:ident $ty:ident;)*) => { match task.tag { - $(task_tag::$tag => cast!(fs_async::$ty).run_from_js_thread()?,)* + // SAFETY: §Dispatch — tag identifies pointee. The task frees itself, so it takes the raw pointer. + $(task_tag::$tag => unsafe { fs_async::$ty::run_from_js_thread(cast_ptr!(fs_async::$ty)) }?,)* // SAFETY: outer arm guard proves one of the table tags matched. _ => unsafe { core::hint::unreachable_unchecked() }, }}; diff --git a/src/runtime/node/node.classes.ts b/src/runtime/node/node.classes.ts index 03621a52efed..25729cccbac5 100644 --- a/src/runtime/node/node.classes.ts +++ b/src/runtime/node/node.classes.ts @@ -335,6 +335,48 @@ export default [ writevSync: { async: false, fn: "writevSync", length: 3 }, realpathNative: { async: true, fn: "realpathNative", length: 3 }, realpathNativeSync: { async: false, fn: "realpathNativeSync", length: 3 }, + // The callback arm of each async operation: argument 0 is the callback. + accessCb: { fn: "accessCb", length: 4 }, + appendFileCb: { fn: "appendFileCb", length: 5 }, + closeCb: { fn: "closeCb", length: 2 }, + copyFileCb: { fn: "copyFileCb", length: 5 }, + existsCb: { fn: "existsCb", length: 3 }, + chownCb: { fn: "chownCb", length: 5 }, + chmodCb: { fn: "chmodCb", length: 4 }, + fchmodCb: { fn: "fchmodCb", length: 4 }, + fchownCb: { fn: "fchownCb", length: 5 }, + fstatCb: { fn: "fstatCb", length: 2 }, + fsyncCb: { fn: "fsyncCb", length: 3 }, + ftruncateCb: { fn: "ftruncateCb", length: 2 }, + futimesCb: { fn: "futimesCb", length: 5 }, + lchmodCb: { fn: "lchmodCb", length: 4 }, + lchownCb: { fn: "lchownCb", length: 5 }, + linkCb: { fn: "linkCb", length: 4 }, + lstatCb: { fn: "lstatCb", length: 2 }, + mkdirCb: { fn: "mkdirCb", length: 4 }, + mkdtempCb: { fn: "mkdtempCb", length: 4 }, + openCb: { fn: "openCb", length: 5 }, + readCb: { fn: "readCb", length: 7 }, + writeCb: { fn: "writeCb", length: 7 }, + readFileCb: { fn: "readFileCb", length: 4 }, + writeFileCb: { fn: "writeFileCb", length: 5 }, + readlinkCb: { fn: "readlinkCb", length: 4 }, + rmCb: { fn: "rmCb", length: 4 }, + rmdirCb: { fn: "rmdirCb", length: 4 }, + realpathCb: { fn: "realpathCb", length: 4 }, + realpathNativeCb: { fn: "realpathNativeCb", length: 4 }, + renameCb: { fn: "renameCb", length: 4 }, + statCb: { fn: "statCb", length: 2 }, + statfsCb: { fn: "statfsCb", length: 3 }, + symlinkCb: { fn: "symlinkCb", length: 5 }, + truncateCb: { fn: "truncateCb", length: 4 }, + unlinkCb: { fn: "unlinkCb", length: 3 }, + utimesCb: { fn: "utimesCb", length: 5 }, + lutimesCb: { fn: "lutimesCb", length: 5 }, + writevCb: { fn: "writevCb", length: 5 }, + readvCb: { fn: "readvCb", length: 5 }, + fdatasyncCb: { fn: "fdatasyncCb", length: 3 }, + readdirCb: { fn: "readdirCb", length: 4 }, Dirent: { getter: "getDirent" }, Stats: { getter: "getStats" }, }, diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index f96f3aea348e..57c5aa7cdf66 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -636,7 +636,7 @@ mod _async_tasks { #[cfg(windows)] pub(crate) struct UVFSRequest { - pub(crate) promise: JSPromiseStrong, + pub(crate) completion: FsCompletion, pub args: ThreadIsolated, pub(crate) global_object: bun_ptr::BackRef, pub(crate) req: uv::fs_t, @@ -661,14 +661,16 @@ mod _async_tasks { self.global_object.get() } + /// `callback`: as for [`AsyncFSTask::create`]. pub(crate) fn create( cx: &bun_jsc::JsThread<'_>, binding: &Binding, task_args: ThreadIsolated, vm: &mut VirtualMachine, + callback: Option, ) -> JSValue { let task = Box::new(Self { - promise: JSPromiseStrong::init(cx.global()), + completion: FsCompletion::new(cx.global(), callback), args: task_args, // Sentinel — overwritten by `uv_callback` (or the early-return arms // below) before any read on the JS thread. `Maybe` is @@ -682,7 +684,7 @@ mod _async_tasks { context: cx.context().id(), }); // Transfer ownership to libuv: the box outlives the async request and is - // reclaimed in `destroy()` (run_from_js_thread → scopeguard). `heap::release` + // reclaimed in `destroy()` (from `run_from_js_thread`). `heap::release` // names that hand-off — it is `Box::leak` under the hood; the reclaim // happens in `destroy()`, not in this scope. let task: &mut Self = bun_core::heap::release(task); @@ -849,7 +851,7 @@ mod _async_tasks { .bun_vm() .event_loop_mut() .enqueue_task(bun_jsc::Task::init(task_ptr)); - return task.promise.value(); + return task.completion.value(); } let pos: i64 = args.position.map(|p| p as i64).unwrap_or(-1); let sum: u64 = bufs.iter().map(|b| b.slice().len() as u64).sum(); @@ -897,7 +899,7 @@ mod _async_tasks { _ => unreachable!("UVFSRequest type not implemented"), } - task.promise.value() + task.completion.value() } extern "C" fn uv_callback(req: *mut uv::fs_t) { @@ -938,41 +940,36 @@ mod _async_tasks { .enqueue_task(bun_jsc::Task::init(this_ptr)); } - pub(crate) fn run_from_js_thread(&mut self) -> JsResult<()> { - // SAFETY: self was Box::leak'd in create(); destroy() runs exactly once on scope exit - let _deinit = - scopeguard::guard(core::ptr::from_mut(self), |p| unsafe { Self::destroy(p) }); - // Move `result` out so the `global_object()` `&self` borrow can coexist - // with consuming it below; the sentinel left behind is dropped in `destroy()`. - let result = core::mem::replace(&mut self.result, Err(sys::Error::default())); - let global_object = self.global_object(); + /// SAFETY: `this` must be the pointer Box::leak'd in `create()`. The body frees it before it calls back, so it is not `&mut self`. + pub(crate) unsafe fn run_from_js_thread(this: *mut Self) -> JsResult<()> { + // SAFETY: fn contract. The borrow ends before `destroy` below. + let (result, completion, global_ref, tracker) = unsafe { + let task = &mut *this; + ( + core::mem::replace(&mut task.result, Err(sys::Error::default())), + core::mem::replace( + &mut task.completion, + FsCompletion::Promise(JSPromiseStrong::empty()), + ), + task.global_object, + task.tracker, + ) + }; + // The arguments pin their buffers. A callback runs inside `resolve`/`reject`, and a pinned buffer cannot be transferred. + // SAFETY: fn contract. `this` is not used again. + unsafe { Self::destroy(this) }; + + let global_object = global_ref.get(); let success = matches!(result, Ok(_)); - let promise_value = self.promise.value(); - let promise = self.promise.get(); - let result = match result { - Err(err) => match err.to_js_with_async_stack(global_object, promise) { - Ok(v) => v, - Err(e) => { - return promise.reject(global_object, Err(e)); - } - }, - Ok(res) => match FsReturn::fs_to_js(res, global_object) { - Ok(v) => v, - Err(e) => { - return promise.reject(global_object, Err(e)); - } - }, + let converted = match result { + Err(err) => completion.error_to_js(global_object, &err), + Ok(res) => FsReturn::fs_to_js(res, global_object), }; - promise_value.ensure_still_alive(); + completion.ensure_still_alive(); - let _dispatch = self.tracker.dispatch(global_object); + let _dispatch = tracker.dispatch(global_object); - if success { - promise.resolve(global_object, result)?; - } else { - promise.reject(global_object, Ok(result))?; - } - Ok(()) + completion.settle(global_object, converted, success) } /// SAFETY: `this` must be the pointer Box::leak'd in `create()`; called exactly once. @@ -1250,10 +1247,118 @@ mod _async_tasks { { } + /// How an async fs operation reports its result: a promise, or the `node:fs` callback, called from the completion. + #[derive(bun_jsc::JsAffine)] + pub(crate) enum FsCompletion { + Promise(JSPromiseStrong), + Callback(bun_jsc::Strong), + } + + impl FsCompletion { + pub(crate) fn new(global: &JSGlobalObject, callback: Option) -> Self { + match callback { + Some(callback) => Self::Callback(bun_jsc::Strong::create( + callback.with_async_context_if_needed(global), + global, + )), + None => Self::Promise(JSPromiseStrong::init(global)), + } + } + + /// What the binding returns to its JS caller. + pub(crate) fn value(&self) -> JSValue { + match self { + Self::Promise(promise) => promise.value(), + Self::Callback(_) => JSValue::UNDEFINED, + } + } + + pub(crate) fn ensure_still_alive(&self) { + match self { + Self::Promise(promise) => promise.value().ensure_still_alive(), + Self::Callback(callback) => callback.get().ensure_still_alive(), + } + } + + pub(crate) fn error_to_js( + &self, + global: &JSGlobalObject, + err: &sys::Error, + ) -> JsResult { + match self { + Self::Promise(promise) => err.to_js_with_async_stack(global, promise.get()), + Self::Callback(_) => err.to_js(global), + } + } + + pub(crate) fn resolve(&self, global: &JSGlobalObject, value: JSValue) -> JsResult<()> { + match self { + Self::Promise(promise) => promise.get().resolve(global, value), + Self::Callback(callback) => { + Self::call(global, callback.get(), &[JSValue::NULL, value]); + Ok(()) + } + } + } + + /// `Err` is the exception a failed conversion left pending, as for `JSPromise::reject`. + pub(crate) fn reject( + &self, + global: &JSGlobalObject, + value: JsResult, + ) -> JsResult<()> { + match self { + Self::Promise(promise) => promise.get().reject(global, value), + // The error value is built exactly as `JSPromise::reject` builds it. + Self::Callback(callback) => { + let error = match value { + Ok(error) => error, + Err(bun_jsc::JsError::OutOfMemory) => global.create_out_of_memory_error(), + Err(bun_jsc::JsError::Terminated) => { + return Err(bun_jsc::JsError::Terminated); + } + Err(bun_jsc::JsError::Thrown) => { + let exception = global.take_exception(bun_jsc::JsError::Thrown); + if exception.is_termination_exception() { + return Err(bun_jsc::top_exception_scope::thrown(global)); + } + exception.to_error().unwrap_or(exception) + } + }; + Self::call(global, callback.get(), &[error]); + Ok(()) + } + } + } + + /// Reports a converted result. `success` says that `converted` holds the value of the operation, not its error. + pub(crate) fn settle( + &self, + global: &JSGlobalObject, + converted: JsResult, + success: bool, + ) -> JsResult<()> { + match converted { + Ok(value) if success => self.resolve(global, value), + error => self.reject(global, error), + } + } + + fn call(global: &JSGlobalObject, callback: JSValue, arguments: &[JSValue]) { + global.bun_vm().event_loop_mut().run_callback( + bun_event_loop::ContextId::NONE, + callback, + global, + JSValue::UNDEFINED, + arguments, + ); + } + } + /// The JS-thread half of an async fs operation. #[derive(bun_jsc::JsAffine)] pub(crate) struct AsyncFSJs { - pub(crate) promise: JSPromiseStrong, + pub(crate) completion: FsCompletion, pub(crate) tracker: AsyncTaskTracker, } @@ -1283,38 +1388,28 @@ mod _async_tasks { let _dispatch = js.tracker.dispatch(global_object); let success = this.result.is_ok(); - let promise_value = js.promise.value(); - let promise = js.promise.get(); - let result = match core::mem::replace(&mut this.result, Err(sys::Error::default())) { - Err(err) => match err.to_js_with_async_stack(global_object, promise) { - Ok(v) => v, - Err(e) => { - return promise.reject(global_object, Err(e)); - } - }, - Ok(res) => match FsReturn::fs_to_js(res, global_object) { - Ok(v) => v, - Err(e) => { - return promise.reject(global_object, Err(e)); - } - }, + let completion = &js.completion; + let converted = match core::mem::replace(&mut this.result, Err(sys::Error::default())) { + Err(err) => completion.error_to_js(global_object, &err), + Ok(res) => FsReturn::fs_to_js(res, global_object), }; - promise_value.ensure_still_alive(); - - if Self::HAVE_ABORT_SIGNAL { - if let Some(signal) = this.args.signal() { - if let Some(abort_error) = signal.node_abort_error_if_aborted(global_object) { - return promise.reject(global_object, Ok(abort_error)); - } - } - } + completion.ensure_still_alive(); - if success { - promise.resolve(global_object, result)?; + let aborted = if Self::HAVE_ABORT_SIGNAL && converted.is_ok() { + this.args + .signal() + .and_then(|signal| signal.node_abort_error_if_aborted(global_object)) } else { - promise.reject(global_object, Ok(result))?; + None + }; + + // The arguments pin their buffers. A callback runs inside `resolve`/`reject`, and a pinned buffer cannot be transferred. + drop(this); + + match aborted { + Some(abort_error) => completion.reject(global_object, Ok(abort_error)), + None => completion.settle(global_object, converted, success), } - Ok(()) } } @@ -1330,16 +1425,18 @@ mod _async_tasks { /// the functions to check .signal.aborted() for early returns. pub(crate) const HAVE_ABORT_SIGNAL: bool = A::HAVE_ABORT_SIGNAL; + /// `callback`: the `node:fs` callback to call with the result. `None` returns a promise. pub(crate) fn create( cx: &bun_jsc::JsThread<'_>, _binding: &Binding, args: ThreadIsolated, vm: &mut VirtualMachine, + callback: Option, ) -> JSValue { let tracker = AsyncTaskTracker::init(vm); tracker.did_schedule(cx.global()); - let promise = JSPromiseStrong::init(cx.global()); - let value = promise.value(); + let completion = FsCompletion::new(cx.global(), callback); + let value = completion.value(); bun_jsc::Job::::schedule( cx, Self { @@ -1348,7 +1445,10 @@ mod _async_tasks { // may be niche-optimised; never construct an all-zero `Result`. result: Err(sys::Error::default()), }, - AsyncFSJs { promise, tracker }, + AsyncFSJs { + completion, + tracker, + }, ); value } @@ -2237,15 +2337,9 @@ mod _async_tasks { ) -> bun_jsc::JsResult<()> { let global_object = cx.global(); let success = this.pending_err.is_none(); - let promise_value = js.promise.value(); - let promise = js.promise.get(); - let result = if let Some(err) = &mut this.pending_err { - match err.to_js_with_async_stack(global_object, promise) { - Ok(v) => v, - Err(e) => { - return promise.reject(global_object, Err(e)); - } - } + let completion = &js.completion; + let converted = if let Some(err) = &this.pending_err { + completion.error_to_js(global_object, err) } else { let res = match core::mem::replace( &mut this.result_list, @@ -2257,22 +2351,12 @@ mod _async_tasks { ResultListEntryValue::Buffers(v) => ret::Readdir::Buffers(v.into_boxed_slice()), ResultListEntryValue::Files(v) => ret::Readdir::Files(v.into_boxed_slice()), }; - match res.to_js(global_object) { - Ok(v) => v, - Err(e) => { - return promise.reject(global_object, Err(e)); - } - } + res.to_js(global_object) }; - promise_value.ensure_still_alive(); + completion.ensure_still_alive(); let _dispatch = js.tracker.dispatch(global_object); drop(this); - if success { - promise.resolve(global_object, result)?; - } else { - promise.reject(global_object, Ok(result))?; - } - Ok(()) + completion.settle(global_object, converted, success) } } @@ -2362,10 +2446,12 @@ mod _async_tasks { }); } + /// `callback`: as for [`AsyncFSTask::create`]. pub(crate) fn create( cx: &bun_jsc::JsThread<'_>, args: ThreadIsolated>, vm: &mut VirtualMachine, + callback: Option, ) -> JSValue { let tag = args.tag(); let encoding = args.encoding; @@ -2385,8 +2471,8 @@ mod _async_tasks { }; let tracker = AsyncTaskTracker::init(vm); tracker.did_schedule(cx.global()); - let promise = JSPromiseStrong::init(cx.global()); - let value = promise.value(); + let completion = FsCompletion::new(cx.global(), callback); + let value = completion.value(); bun_jsc::Job::::schedule( cx, AsyncReaddirRecursiveTask { @@ -2405,7 +2491,10 @@ mod _async_tasks { pending_err: None, pending_err_mutex: bun_threading::Mutex::default(), }, - AsyncFSJs { promise, tracker }, + AsyncFSJs { + completion, + tracker, + }, ); value } diff --git a/src/runtime/node/node_fs_binding.rs b/src/runtime/node/node_fs_binding.rs index 2d7875b47042..5696b42e8d77 100644 --- a/src/runtime/node/node_fs_binding.rs +++ b/src/runtime/node/node_fs_binding.rs @@ -50,6 +50,31 @@ where } } +/// Which arm of an async binding was called. +#[derive(Clone, Copy)] +enum AsyncArm { + /// `node:fs/promises`: returns a promise. + Promise, + /// Argument 0 is the callback. `fs.promises` forwards user arguments, so a trailing function cannot mark this arm. + Callback, +} + +/// The arguments of the operation, and the callback for [`AsyncArm::Callback`]. +fn split_callback<'a>( + global: &JSGlobalObject, + frame: &'a CallFrame, + arm: AsyncArm, +) -> JsResult<(Option, &'a [JSValue])> { + let arguments = frame.arguments(); + match arm { + AsyncArm::Promise => Ok((None, arguments)), + AsyncArm::Callback => match arguments.split_first() { + Some((callback, rest)) if callback.is_callable() => Ok((Some(*callback), rest)), + _ => Err(global.throw_invalid_arguments(format_args!("callback must be a function"))), + }, + } +} + /// `Bindings(FunctionEnum).runAsync` for every operation except `.cp` / /// `.readdir` (those have bespoke entry points below). /// @@ -61,16 +86,19 @@ fn run_async( this: &Binding, global: &JSGlobalObject, frame: &CallFrame, + arm: AsyncArm, create_task: fn( &bun_jsc::JsThread<'_>, &Binding, ThreadIsolated, &mut VirtualMachine, + Option, ) -> JSValue, ) -> JsResult { - let args = match parse_async_args::(global, frame) { - Ok(args) => args, - Err(result) => return result, + let (callback, arguments) = split_callback(global, frame, arm)?; + let args = match parse_async_args::(global, arguments)? { + ParsedAsyncArgs::Args(args) => args, + ParsedAsyncArgs::Rejected(error) => return reject_before_schedule(global, callback, error), }; let vm: &mut VirtualMachine = global.bun_vm().as_mut(); Ok(create_task( @@ -78,34 +106,52 @@ fn run_async( this, args, vm, + callback, )) } -/// Parses a promise-returning binding's arguments; `Err` is what the binding returns instead. +enum ParsedAsyncArgs { + Args(ThreadIsolated), + /// The operation fails without being scheduled (an aborted signal, a path that is too long). + Rejected(JSValue), +} + +/// Parses an async binding's arguments. A validation error is thrown, as node does for both arms. fn parse_async_args( global: &JSGlobalObject, - frame: &CallFrame, -) -> Result, JsResult> { + arguments: &[JSValue], +) -> JsResult> { let vm: &VirtualMachine = global.bun_vm(); - let mut slice = ArgumentsSlice::init(vm, frame.arguments()); - let args = A::from_js_async(global, &mut slice).map_err(Err)?; - - let rejection = 'rejection: { - if A::HAVE_ABORT_SIGNAL { - if let Some(abort_error) = args - .signal() - .and_then(|signal| signal.node_abort_error_if_aborted(global)) - { - break 'rejection abort_error; - } - } - if let Some(err) = slice.deferred_error.take() { - break 'rejection (*err).to_error_instance(global); + let mut slice = ArgumentsSlice::init(vm, arguments); + let args = A::from_js_async(global, &mut slice)?; + + if A::HAVE_ABORT_SIGNAL { + if let Some(abort_error) = args + .signal() + .and_then(|signal| signal.node_abort_error_if_aborted(global)) + { + return Ok(ParsedAsyncArgs::Rejected(abort_error)); } - return Ok(args); - }; + } + if let Some(err) = slice.deferred_error.take() { + return Ok(ParsedAsyncArgs::Rejected((*err).to_error_instance(global))); + } + Ok(ParsedAsyncArgs::Args(args)) +} - Err(Ok(JSPromise::rejected_promise(global, rejection).to_js())) +/// An error found before scheduling. The callback goes to nextTick bare: it asserts against an `AsyncContextFrame`. +fn reject_before_schedule( + global: &JSGlobalObject, + callback: Option, + error: JSValue, +) -> JsResult { + match callback { + None => Ok(JSPromise::rejected_promise(global, error).to_js()), + Some(callback) => { + JSValue::call_next_tick_1(callback, global, error)?; + Ok(JSValue::UNDEFINED) + } + } } #[inline(always)] @@ -167,9 +213,9 @@ impl Binding { /// `callAsync(.cp)`. pub(crate) fn cp(this: &Self, global: &JSGlobalObject, frame: &CallFrame) -> JsResult { - let cp_args = match parse_async_args::>(global, frame) { - Ok(args) => args, - Err(result) => return result, + let cp_args = match parse_async_args::>(global, frame.arguments())? { + ParsedAsyncArgs::Args(args) => args, + ParsedAsyncArgs::Rejected(error) => return reject_before_schedule(global, None, error), }; let vm: &mut VirtualMachine = global.bun_vm().as_mut(); Ok(AsyncCpTask::create( @@ -207,9 +253,29 @@ impl Binding { global: &JSGlobalObject, frame: &CallFrame, ) -> JsResult { - let rd_args = match parse_async_args::>(global, frame) { - Ok(args) => args, - Err(result) => return result, + Self::run_readdir(this, global, frame, AsyncArm::Promise) + } + + pub(crate) fn readdir_cb( + this: &Self, + global: &JSGlobalObject, + frame: &CallFrame, + ) -> JsResult { + Self::run_readdir(this, global, frame, AsyncArm::Callback) + } + + fn run_readdir( + this: &Self, + global: &JSGlobalObject, + frame: &CallFrame, + arm: AsyncArm, + ) -> JsResult { + let (callback, arguments) = split_callback(global, frame, arm)?; + let rd_args = match parse_async_args::>(global, arguments)? { + ParsedAsyncArgs::Args(args) => args, + ParsedAsyncArgs::Rejected(error) => { + return reject_before_schedule(global, callback, error); + } }; let vm: &mut VirtualMachine = global.bun_vm().as_mut(); // /$bunfs/ is in-memory; readdir_inner handles it (recursive included). @@ -220,6 +286,7 @@ impl Binding { &global.js_thread_of_caller(frame), rd_args, vm, + callback, )); } Ok(async_::Readdir::create( @@ -227,6 +294,7 @@ impl Binding { this, rd_args, vm, + callback, )) } @@ -285,7 +353,7 @@ impl Binding { /// Generates the `pub const = call{Async,Sync}(.)` block. /// Each row supplies the `(args, ret, NodeFSFunctionEnum)` triple for one op. macro_rules! node_fs_bindings { - ( $( $sync:ident / $async_:ident => $F:ident, $Args:ty, $Ret:ty ; )* ) => { + ( $( $sync:ident / $async_:ident / $callback:ident => $F:ident, $Args:ty, $Ret:ty ; )* ) => { impl Binding { $( pub(crate) const $sync: NodeFSFunction = @@ -295,7 +363,14 @@ macro_rules! node_fs_bindings { global: &JSGlobalObject, frame: &CallFrame, ) -> JsResult { - run_async::<$Args>(this, global, frame, async_::$F::create) + run_async::<$Args>(this, global, frame, AsyncArm::Promise, async_::$F::create) + } + pub(crate) fn $callback( + this: &Self, + global: &JSGlobalObject, + frame: &CallFrame, + ) -> JsResult { + run_async::<$Args>(this, global, frame, AsyncArm::Callback, async_::$F::create) } )* } @@ -304,46 +379,46 @@ macro_rules! node_fs_bindings { #[rustfmt::skip] node_fs_bindings! { - access_sync / access => Access, args::Access<'static>, ret::Access; - append_file_sync / append_file => AppendFile, args::AppendFile<'static>, ret::AppendFile; - close_sync / close => Close, args::Close, ret::Close; - copy_file_sync / copy_file => CopyFile, args::CopyFile<'static>, ret::CopyFile; - exists_sync / exists => Exists, args::Exists<'static>, ret::Exists; - chown_sync / chown => Chown, args::Chown<'static>, ret::Chown; - chmod_sync / chmod => Chmod, args::Chmod<'static>, ret::Chmod; - fchmod_sync / fchmod => Fchmod, args::FChmod, ret::Fchmod; - fchown_sync / fchown => Fchown, args::Fchown, ret::Fchown; - fstat_sync / fstat => Fstat, args::Fstat, ret::Fstat; - fsync_sync / fsync => Fsync, args::Fsync, ret::Fsync; - ftruncate_sync / ftruncate => Ftruncate, args::FTruncate, ret::Ftruncate; - futimes_sync / futimes => Futimes, args::Futimes, ret::Futimes; - lchmod_sync / lchmod => Lchmod, args::LCHmod<'static>, ret::Lchmod; - lchown_sync / lchown => Lchown, args::LChown<'static>, ret::Lchown; - link_sync / link => Link, args::Link<'static>, ret::Link; - lstat_sync / lstat => Lstat, args::Lstat<'static>, ret::Lstat; - mkdir_sync / mkdir => Mkdir, args::Mkdir<'static>, ret::Mkdir; - mkdtemp_sync / mkdtemp => Mkdtemp, args::MkdirTemp<'static>, ret::Mkdtemp; - open_sync / open => Open, args::Open<'static>, ret::Open; - read_sync / read => Read, args::Read, ret::Read; - write_sync / write => Write, args::Write<'static>, ret::Write; - read_file_sync / read_file => ReadFile, args::ReadFile<'static>, ret::ReadFile; - write_file_sync / write_file => WriteFile, args::WriteFile<'static>, ret::WriteFile; - readlink_sync / readlink => Readlink, args::Readlink<'static>, ret::Readlink; - rm_sync / rm => Rm, args::Rm<'static>, ret::Rm; - rmdir_sync / rmdir => Rmdir, args::RmDir<'static>, ret::Rmdir; - realpath_sync / realpath => RealpathNonNative, args::Realpath<'static>, ret::Realpath; - realpath_native_sync / realpath_native => Realpath, args::Realpath<'static>, ret::Realpath; - rename_sync / rename => Rename, args::Rename<'static>, ret::Rename; - stat_sync / stat => Stat, args::Stat<'static>, ret::Stat; - statfs_sync / statfs => Statfs, args::StatFS<'static>, ret::StatFS; - symlink_sync / symlink => Symlink, args::Symlink<'static>, ret::Symlink; - truncate_sync / truncate => Truncate, args::Truncate<'static>, ret::Truncate; - unlink_sync / unlink => Unlink, args::Unlink<'static>, ret::Unlink; - utimes_sync / utimes => Utimes, args::Utimes<'static>, ret::Utimes; - lutimes_sync / lutimes => Lutimes, args::Lutimes<'static>, ret::Lutimes; - writev_sync / writev => Writev, args::Writev, ret::Writev; - readv_sync / readv => Readv, args::Readv, ret::Readv; - fdatasync_sync / fdatasync => Fdatasync, args::FdataSync, ret::Fdatasync; + access_sync / access / access_cb => Access, args::Access<'static>, ret::Access; + append_file_sync / append_file / append_file_cb => AppendFile, args::AppendFile<'static>, ret::AppendFile; + close_sync / close / close_cb => Close, args::Close, ret::Close; + copy_file_sync / copy_file / copy_file_cb => CopyFile, args::CopyFile<'static>, ret::CopyFile; + exists_sync / exists / exists_cb => Exists, args::Exists<'static>, ret::Exists; + chown_sync / chown / chown_cb => Chown, args::Chown<'static>, ret::Chown; + chmod_sync / chmod / chmod_cb => Chmod, args::Chmod<'static>, ret::Chmod; + fchmod_sync / fchmod / fchmod_cb => Fchmod, args::FChmod, ret::Fchmod; + fchown_sync / fchown / fchown_cb => Fchown, args::Fchown, ret::Fchown; + fstat_sync / fstat / fstat_cb => Fstat, args::Fstat, ret::Fstat; + fsync_sync / fsync / fsync_cb => Fsync, args::Fsync, ret::Fsync; + ftruncate_sync / ftruncate / ftruncate_cb => Ftruncate, args::FTruncate, ret::Ftruncate; + futimes_sync / futimes / futimes_cb => Futimes, args::Futimes, ret::Futimes; + lchmod_sync / lchmod / lchmod_cb => Lchmod, args::LCHmod<'static>, ret::Lchmod; + lchown_sync / lchown / lchown_cb => Lchown, args::LChown<'static>, ret::Lchown; + link_sync / link / link_cb => Link, args::Link<'static>, ret::Link; + lstat_sync / lstat / lstat_cb => Lstat, args::Lstat<'static>, ret::Lstat; + mkdir_sync / mkdir / mkdir_cb => Mkdir, args::Mkdir<'static>, ret::Mkdir; + mkdtemp_sync / mkdtemp / mkdtemp_cb => Mkdtemp, args::MkdirTemp<'static>, ret::Mkdtemp; + open_sync / open / open_cb => Open, args::Open<'static>, ret::Open; + read_sync / read / read_cb => Read, args::Read, ret::Read; + write_sync / write / write_cb => Write, args::Write<'static>, ret::Write; + read_file_sync / read_file / read_file_cb => ReadFile, args::ReadFile<'static>, ret::ReadFile; + write_file_sync / write_file / write_file_cb => WriteFile, args::WriteFile<'static>, ret::WriteFile; + readlink_sync / readlink / readlink_cb => Readlink, args::Readlink<'static>, ret::Readlink; + rm_sync / rm / rm_cb => Rm, args::Rm<'static>, ret::Rm; + rmdir_sync / rmdir / rmdir_cb => Rmdir, args::RmDir<'static>, ret::Rmdir; + realpath_sync / realpath / realpath_cb => RealpathNonNative, args::Realpath<'static>, ret::Realpath; + realpath_native_sync / realpath_native / realpath_native_cb => Realpath, args::Realpath<'static>, ret::Realpath; + rename_sync / rename / rename_cb => Rename, args::Rename<'static>, ret::Rename; + stat_sync / stat / stat_cb => Stat, args::Stat<'static>, ret::Stat; + statfs_sync / statfs / statfs_cb => Statfs, args::StatFS<'static>, ret::StatFS; + symlink_sync / symlink / symlink_cb => Symlink, args::Symlink<'static>, ret::Symlink; + truncate_sync / truncate / truncate_cb => Truncate, args::Truncate<'static>, ret::Truncate; + unlink_sync / unlink / unlink_cb => Unlink, args::Unlink<'static>, ret::Unlink; + utimes_sync / utimes / utimes_cb => Utimes, args::Utimes<'static>, ret::Utimes; + lutimes_sync / lutimes / lutimes_cb => Lutimes, args::Lutimes<'static>, ret::Lutimes; + writev_sync / writev / writev_cb => Writev, args::Writev, ret::Writev; + readv_sync / readv / readv_cb => Readv, args::Readv, ret::Readv; + fdatasync_sync / fdatasync / fdatasync_cb => Fdatasync, args::FdataSync, ret::Fdatasync; } // `readdirSync` goes through the generic sync path; only the async side is diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index a1b0da7e718b..54b8c275cdf7 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -2048,6 +2048,7 @@ impl BlobExt for Blob { binding, crate::node::fs::args::Stat::owned(path_like.slice().to_vec()), vm, + None, )) } PathOrFileDescriptor::Fd(fd) => { @@ -2062,6 +2063,7 @@ impl BlobExt for Blob { binding, crate::node::fs::args::Fstat::for_fd(*fd), vm, + None, )) } } diff --git a/src/runtime/webcore/blob/Store.rs b/src/runtime/webcore/blob/Store.rs index 30b35cb77fe5..33a6aaa89aba 100644 --- a/src/runtime/webcore/blob/Store.rs +++ b/src/runtime/webcore/blob/Store.rs @@ -230,6 +230,7 @@ impl FileExt for File { &binding, node_fs::args::Unlink::owned(path_like.slice().to_vec()), cx.vm().as_mut(), + None, )) } PathOrFileDescriptor::Fd(_) => Ok(JSPromise::resolved_promise_value( diff --git a/test/js/node/fs/fs.test.ts b/test/js/node/fs/fs.test.ts index 55663f10f4e3..b52ab508b112 100644 --- a/test/js/node/fs/fs.test.ts +++ b/test/js/node/fs/fs.test.ts @@ -7129,6 +7129,201 @@ describe("a throw from a node-style callback is an uncaughtException", () => { }); }); +// Node calls an fs callback from the event loop, then runs the process.nextTick +// queue, then the microtasks. +describe("a process.nextTick queued by an fs callback runs before a microtask it queues", () => { + type Start = (dir: string, callback: () => void) => void; + + // The first process.nextTick() of a process runs right after the microtask + // that queued it, which hides a callback that runs inside a microtask. + function tickQueueExists() { + return new Promise(resolve => process.nextTick(resolve)); + } + + function withFd(name: string, flags: string, start: (fd: number, callback: () => void) => void): Start { + return (dir, callback) => { + const fd = openSync(join(dir, name), flags); + start(fd, () => { + closeSync(fd); + callback(); + }); + }; + } + + const cases: Array<[string, Start]> = [ + ["access", (dir, cb) => fs.access(join(dir, "file.txt"), cb)], + ["appendFile", (dir, cb) => fs.appendFile(join(dir, "file.txt"), "x", cb)], + ["chmod", (dir, cb) => fs.chmod(join(dir, "file.txt"), 0o644, cb)], + ["close", (dir, cb) => fs.close(openSync(join(dir, "file.txt"), "r"), cb)], + ["copyFile", (dir, cb) => fs.copyFile(join(dir, "file.txt"), join(dir, "copy.txt"), cb)], + ["cp", (dir, cb) => fs.cp(join(dir, "file.txt"), join(dir, "cp.txt"), cb)], + ["exists", (dir, cb) => fs.exists(join(dir, "file.txt"), cb)], + ["exists (missing)", (dir, cb) => fs.exists(join(dir, "missing"), cb)], + ["fdatasync", withFd("file.txt", "r+", (fd, cb) => fs.fdatasync(fd, cb))], + ["fstat", withFd("file.txt", "r", (fd, cb) => fs.fstat(fd, cb))], + ["fsync", withFd("file.txt", "r+", (fd, cb) => fs.fsync(fd, cb))], + ["ftruncate", withFd("file.txt", "r+", (fd, cb) => fs.ftruncate(fd, 1, cb))], + ["futimes", withFd("file.txt", "r+", (fd, cb) => fs.futimes(fd, 1, 1, cb))], + ["lstat", (dir, cb) => fs.lstat(join(dir, "file.txt"), cb)], + ["lutimes", (dir, cb) => fs.lutimes(join(dir, "file.txt"), 1, 1, cb)], + ["mkdir", (dir, cb) => fs.mkdir(join(dir, "a", "b"), { recursive: true }, cb)], + ["mkdtemp", (dir, cb) => fs.mkdtemp(join(dir, "tmp-"), cb)], + [ + "open", + (dir, cb) => + fs.open(join(dir, "file.txt"), "r", (err, fd) => { + closeSync(fd); + cb(); + }), + ], + ["read", withFd("file.txt", "r", (fd, cb) => fs.read(fd, Buffer.alloc(4), 0, 4, 0, cb))], + ["readv", withFd("file.txt", "r", (fd, cb) => fs.readv(fd, [Buffer.alloc(4)], 0, cb))], + ["readdir", (dir, cb) => fs.readdir(dir, cb)], + ["readdir (recursive)", (dir, cb) => fs.readdir(dir, { recursive: true }, cb)], + ["readFile", (dir, cb) => fs.readFile(join(dir, "file.txt"), cb)], + ["realpath", (dir, cb) => fs.realpath(join(dir, "file.txt"), cb)], + ["realpath.native", (dir, cb) => fs.realpath.native(join(dir, "file.txt"), cb)], + ["rename", (dir, cb) => fs.rename(join(dir, "file.txt"), join(dir, "renamed.txt"), cb)], + ["rm", (dir, cb) => fs.rm(join(dir, "file.txt"), cb)], + ["rmdir", (dir, cb) => fs.rmdir(join(dir, "empty"), cb)], + ["stat", (dir, cb) => fs.stat(join(dir, "file.txt"), cb)], + ["stat (missing)", (dir, cb) => fs.stat(join(dir, "missing"), cb)], + ["statfs", (dir, cb) => fs.statfs(dir, cb)], + ["truncate", (dir, cb) => fs.truncate(join(dir, "file.txt"), 1, cb)], + ["unlink", (dir, cb) => fs.unlink(join(dir, "file.txt"), cb)], + ["utimes", (dir, cb) => fs.utimes(join(dir, "file.txt"), 1, 1, cb)], + ["write", withFd("file.txt", "r+", (fd, cb) => fs.write(fd, Buffer.from("x"), 0, 1, 0, cb))], + ["write (string)", withFd("file.txt", "r+", (fd, cb) => fs.write(fd, "x", 0, "utf8", cb))], + ["writev", withFd("file.txt", "r+", (fd, cb) => fs.writev(fd, [Buffer.from("x")], 0, cb))], + ["writeFile", (dir, cb) => fs.writeFile(join(dir, "written.txt"), "x", cb)], + [ + "Dir.read", + (dir, cb) => { + const handle = fs.opendirSync(dir); + handle.read(() => { + cb(); + handle.close(() => {}); + }); + }, + ], + ["Dir.close", (dir, cb) => fs.opendirSync(dir).close(cb)], + ]; + if (!isWindows) { + cases.push( + ["link", (dir, cb) => fs.link(join(dir, "file.txt"), join(dir, "hardlink"), cb)], + ["readlink", (dir, cb) => fs.readlink(join(dir, "symlink"), cb)], + ["symlink", (dir, cb) => fs.symlink(join(dir, "file.txt"), join(dir, "symlink-2"), cb)], + ); + } + + it.each(cases)("%s", async (_name, start) => { + using dir = tempDir("fs-callback-order", { "file.txt": "hello", "empty": {} }); + if (!isWindows) symlinkSync(join(String(dir), "file.txt"), join(String(dir), "symlink")); + await tickQueueExists(); + + const { promise, resolve } = Promise.withResolvers(); + const order: string[] = []; + const step = (name: string) => () => { + order.push(name); + if (order.length === 3) resolve(order); + }; + start(String(dir), () => { + step("callback")(); + process.nextTick(step("nextTick")); + queueMicrotask(step("microtask")); + }); + + expect(await promise).toEqual(["callback", "nextTick", "microtask"]); + }); + + it("and both run before the callback of the next operation", async () => { + using dir = tempDir("fs-callback-order", { "file.txt": "hello" }); + await tickQueueExists(); + + const { promise, resolve } = Promise.withResolvers(); + const order: string[] = []; + let callbacks = 0; + for (let i = 0; i < 3; i++) { + fs.stat(join(String(dir), "file.txt"), () => { + const n = callbacks++; + order.push(`callback ${n}`); + process.nextTick(() => order.push(`nextTick ${n}`)); + queueMicrotask(() => { + order.push(`microtask ${n}`); + if (n === 2) resolve(order); + }); + }); + } + + expect(await promise).toEqual([ + "callback 0", + "nextTick 0", + "microtask 0", + "callback 1", + "nextTick 1", + "microtask 1", + "callback 2", + "nextTick 2", + "microtask 2", + ]); + }); + + // In a subprocess: the replacement must not reach the other tests, and a held + // callback ends the process with no output instead of a hang. + it("while user code has replaced process.nextTick", async () => { + using dir = tempDir("fs-callback-order", { "file.txt": "hello" }); + const script = ` + const fs = require("fs"); + const realNextTick = process.nextTick; + const order = []; + process.on("exit", () => console.log(order.join(" "))); + // The tick queue exists before the operation starts. + realNextTick(() => { + // What a fake-timer library does: hold the job until the fake clock runs. + process.nextTick = () => {}; + fs.stat(${JSON.stringify(join(String(dir), "file.txt"))}, () => { + order.push("callback"); + realNextTick(() => order.push("nextTick")); + queueMicrotask(() => order.push("microtask")); + }); + }); + `; + await using proc = Bun.spawn({ cmd: [bunExe(), "-e", script], env: bunEnv, stdout: "pipe", stderr: "pipe" }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toBe(""); + expect(stdout.trim()).toBe("callback nextTick microtask"); + expect(exitCode).toBe(0); + }); +}); + +// The operation pins the buffer while it runs. Node can transfer it from the callback. +describe("the buffer of a finished operation can be transferred from its callback", () => { + type Start = (fd: number, view: Uint8Array, callback: () => void) => void; + const cases: Array<[string, Start]> = [ + ["read", (fd, view, cb) => fs.read(fd, view, 0, view.byteLength, 0, cb)], + ["readv", (fd, view, cb) => fs.readv(fd, [view], 0, cb)], + ["write", (fd, view, cb) => fs.write(fd, view, 0, view.byteLength, 0, cb)], + ["writev", (fd, view, cb) => fs.writev(fd, [view], 0, cb)], + ]; + + it.each(cases)("%s", async (_name, start) => { + using dir = tempDir("fs-callback-transfer", { "file.txt": "hello world" }); + const fd = openSync(join(String(dir), "file.txt"), "r+"); + try { + const buffer = new ArrayBuffer(8); + const { promise, resolve } = Promise.withResolvers<{ source: number; moved: number }>(); + start(fd, new Uint8Array(buffer), () => { + const moved = structuredClone(buffer, { transfer: [buffer] }); + resolve({ source: buffer.byteLength, moved: moved.byteLength }); + }); + expect(await promise).toEqual({ source: 0, moved: 8 }); + } finally { + closeSync(fd); + } + }); +}); + describe("fs.Utf8Stream", () => { // A write started from the reopen 'ready' listener is still in flight when the // reopen path announces 'drain'; the write's own completion is what must emit it.