Skip to content
Open
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
3 changes: 2 additions & 1 deletion src/codegen/builtins/host/runtime-dependencies.builtins.ts
Original file line number Diff line number Diff line change
Expand Up @@ -329,7 +329,8 @@ const promises = ["Promise"];
// The overload for a delay that fits a timer says that it runs no code.
const timersWithOtherDelays = ["setInterval", "setTimeout"].map(path => ({ path, when: b.whenCallRunsOtherCode }));

// Each takes its callback first, and queues it with the error where it fails before anything is scheduled.
// Each takes its callback first, and queues it with the error where it fails before anything is scheduled. readFileCb,
// writeFileCb and appendFileCb return the error of a signal that is already aborted instead, and do not queue the callback.
const fsMethodsTakingCallback = [
"access",
"appendFile",
Expand Down
6 changes: 5 additions & 1 deletion src/js/internal/trace_events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -654,8 +654,9 @@ function wrapFsCallbackMethod(original: Wrapped, names: string[]) {
return function (this: unknown, callback: Wrapped, ...args: unknown[]) {
if (suppressFsEvents || !isCategoryGroupEnabled(kFsAsyncCat)) return original.$apply(this, arguments);
for (let i = 0; i < names.length; i++) emitEvent("b", kFsAsyncCat, names[i]);
let aborted;
try {
return original.$call(
aborted = original.$call(
this,
function () {
emitFsAsyncEnd(names);
Expand All @@ -667,6 +668,9 @@ function wrapFsCallbackMethod(original: Wrapped, names: string[]) {
emitFsAsyncEnd(names);
throw err;
}
// The binding returns the error of a signal that is already aborted, and then nothing calls back.
if (aborted !== undefined) emitFsAsyncEnd(names);
return aborted;
};
}

Expand Down
20 changes: 11 additions & 9 deletions src/js/node/fs.ts
Original file line number Diff line number Diff line change
Expand Up @@ -84,9 +84,9 @@ var access = function access(path: PathLike, mode?: number | Callback, callback?
options = undefined;
}

callback = ensureCallback(callback);

fs.appendFileCb(callback, path, data, options);
// As in readFile: the binding returns the error of a signal that is already aborted.
const aborted = fs.appendFileCb(ensureCallback(callback), path, data, options);
if (aborted !== undefined) callback(aborted);
},
close = function close(fd: number, callback?: Callback) {
if ($isCallable(callback)) {
Expand Down Expand Up @@ -364,15 +364,17 @@ var access = function access(path: PathLike, mode?: number | Callback, callback?
},
readFile = function readFile(path, options, callback?) {
callback ||= options;
callback = ensureCallback(callback);

fs.readFileCb(callback, path, options);
// The binding returns the error of a signal that is already aborted, and does not call back. Node calls back
// before it returns, and what the callback throws reaches the caller, so this calls the callback and not its
// guardCallback wrapper. https://github.com/nodejs/node/blob/v26.12.1/lib/fs.js#L380-L386
const aborted = fs.readFileCb(ensureCallback(callback), path, options);
if (aborted !== undefined) callback(aborted);
},
writeFile = function writeFile(path, data, options, callback?) {
callback ||= options;
callback = ensureCallback(callback);

fs.writeFileCb(callback, path, data, options);
// As in readFile: the binding returns the error of a signal that is already aborted.
const aborted = fs.writeFileCb(ensureCallback(callback), path, data, options);
if (aborted !== undefined) callback(aborted);
},
readlink = function readlink(path, options, callback?) {
if ($isCallable(options)) {
Expand Down
10 changes: 8 additions & 2 deletions src/js/private.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1927,12 +1927,18 @@ declare namespace $Node {
writev(fd: number, buffers: readonly AnyView[], position?: number | null): Promise<number>;
}

/** `xCb(callback, ...arguments of x)`. The error is `null` when the operation succeeded. */
/**
* `xCb(callback, ...arguments of x)`. The error is `null` when the operation succeeded.
* `readFileCb`, `writeFileCb` and `appendFileCb` return the error of a signal that is already aborted, and then do not call back.
*/
type NodeJSFSCallbacks = {
[K in Exclude<keyof NodeJSFSPromises, "write"> as `${K}Cb`]: NodeJSFSPromises[K] extends (
...args: infer A
) => Promise<infer R>
? (callback: (error: Error | null, result: R) => unknown, ...args: A) => void
? (
callback: (error: Error | null, result: R) => unknown,
...args: A
) => K extends "appendFile" | "readFile" | "writeFile" ? Error | undefined : void
: never;
};

Expand Down
25 changes: 24 additions & 1 deletion src/runtime/node/node_fs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1365,6 +1365,16 @@ mod _async_tasks {
// NewAsyncFSTask — runs a NodeFS method on the thread pool.
// ──────────────────────────────────────────────────────────────────────────

/// When the callback of an async call hears of an error that is found before the operation is scheduled.
#[derive(Clone, Copy)]
pub(crate) enum EarlyDelivery {
/// Before the call returns, as node's `checkAborted` does.
/// https://github.com/nodejs/node/blob/v26.12.1/lib/fs.js#L380-L386
BeforeReturn,
/// Node gets the error from the thread pool (a path that is too long), or defers the whole call (`fs.readFile(fd)`).
NextTick,
}

/// One `fs.*` operation's parsed arguments.
pub(crate) trait FsArgument: Sized + ThreadIsolatedArg {
const HAVE_ABORT_SIGNAL: bool = false;
Expand All @@ -1387,6 +1397,10 @@ mod _async_tasks {
fn signal(&self) -> Option<&AbortSignal> {
None
}
/// When a callback hears of a signal that is already aborted when the parse ends.
fn aborted_delivery(&self) -> EarlyDelivery {
EarlyDelivery::BeforeReturn
}
/// Runs on the JS thread before the result reaches JS. See [`PinnedArrayBuffer::write_back`].
fn write_back(&mut self, _bytes_read: u64) {}
fn paths(&self) -> [Option<&PathLike<'static>>; 2] {
Expand Down Expand Up @@ -1496,6 +1510,14 @@ mod _async_tasks {
fn signal(&self) -> Option<&AbortSignal> {
self.signal.as_deref()
}
/// Node defers the fd form before it looks at the signal.
/// https://github.com/nodejs/node/blob/v26.12.1/lib/fs.js#L442-L450
fn aborted_delivery(&self) -> EarlyDelivery {
match self.path {
PathOrFileDescriptor::Fd(_) => EarlyDelivery::NextTick,
PathOrFileDescriptor::Path(_) => EarlyDelivery::BeforeReturn,
}
}
fn paths(&self) -> [Option<&PathLike<'static>>; 2] {
[path_of(&self.path), None]
}
Expand Down Expand Up @@ -3113,7 +3135,8 @@ mod _async_tasks {
}
} // mod _async_tasks
pub(crate) use _async_tasks::{
AsyncCpTask, AsyncReaddirRecursiveTask, FsArgument, FsReturn, ShellAsyncCpTask, async_,
AsyncCpTask, AsyncReaddirRecursiveTask, EarlyDelivery, FsArgument, FsReturn, ShellAsyncCpTask,
async_,
};

// ──────────────────────────────────────────────────────────────────────────
Expand Down
9 changes: 8 additions & 1 deletion src/runtime/node/node_fs_binding.bind.ts
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,13 @@ const syncResults: Record<string, readonly string[]> = {
writev: ["number"],
};

/** What `xCb` returns beside `undefined`: the error of a signal that is already aborted, which `fs.ts` calls back with. */
const callbackResults: Record<string, readonly string[]> = {
appendFile: ["object"],
readFile: ["object"],
writeFile: ["object"],
};

const getter = { getter: b.any.returning("object"), capability: "js" } as const;

/** The default export of `src/js/internal/fs/binding.ts`. */
Expand All @@ -114,7 +121,7 @@ export const NodeJSFS = b.class({
[
...Object.entries(operations).flatMap(([name, [length, syncLength]]) => [
[name, { ...operation(length, "Promise"), uses }] as const,
[`${name}Cb`, operation(length + 1, "undefined")] as const,
[`${name}Cb`, operation(length + 1, "undefined", ...(callbackResults[name] ?? []))] as const,
[`${name}Sync`, { ...operation(syncLength, ...(syncResults[name] ?? ["undefined"])), uses }] as const,
]),
["cp", { ...operation(2, "Promise"), uses }] as const,
Expand Down
49 changes: 36 additions & 13 deletions src/runtime/node/node_fs_binding.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,8 @@ use bun_sys_jsc::SystemErrorJsc as _;
use crate::generated_bindgen::node_fs_binding::RmOptions;
use crate::node::ThreadIsolated;
use crate::node::fs::{
self, AsyncCpTask, AsyncReaddirRecursiveTask, Flavor, FsArgument, FsReturn, NodeFS,
NodeFSDispatch, NodeFSFunctionEnum, Op, args, async_, ret,
self, AsyncCpTask, AsyncReaddirRecursiveTask, EarlyDelivery, Flavor, FsArgument, FsReturn,
NodeFS, NodeFSDispatch, NodeFSFunctionEnum, Op, args, async_, ret,
};

// The (`args::*`, `ret::*`, `NodeFS::<method>`, `async_::*`) quadruples are
Expand Down Expand Up @@ -97,7 +97,9 @@ fn run_async<A: FsArgument>(
let (callback, arguments) = split_callback(global, args, arm)?;
let args = match parse_async_args::<A>(global, arguments)? {
ParsedAsyncArgs::Args(args) => args,
ParsedAsyncArgs::Rejected(error) => return reject_before_schedule(global, callback, error),
ParsedAsyncArgs::Rejected { error, delivery } => {
return reject_before_schedule(global, callback, error, delivery);
}
};
let vm: &mut VirtualMachine = global.bun_vm().as_mut();
Ok(create_task(&cx, this, args, vm, callback))
Expand All @@ -106,7 +108,10 @@ fn run_async<A: FsArgument>(
enum ParsedAsyncArgs<A> {
Args(ThreadIsolated<A>),
/// The operation fails without being scheduled (an aborted signal, a path that is too long).
Rejected(JSValue),
Rejected {
error: JSValue,
delivery: EarlyDelivery,
},
}

/// Parses an async binding's arguments. A validation error is thrown, as node does for both arms.
Expand All @@ -123,24 +128,40 @@ fn parse_async_args<A: FsArgument>(
.signal()
.and_then(|signal| signal.node_abort_error_if_aborted(global))
{
return Ok(ParsedAsyncArgs::Rejected(abort_error));
return Ok(ParsedAsyncArgs::Rejected {
error: abort_error,
delivery: args.aborted_delivery(),
});
}
}
if let Some(err) = slice.deferred_error.take() {
return Ok(ParsedAsyncArgs::Rejected((*err).to_error_instance(global)));
return Ok(ParsedAsyncArgs::Rejected {
error: (*err).to_error_instance(global),
delivery: EarlyDelivery::NextTick,
});
}
Ok(ParsedAsyncArgs::Args(args))
}

/// An error found before scheduling. The callback goes to nextTick bare: it asserts against an `AsyncContextFrame`.
/// An error found before scheduling. What the callback arm returns is `undefined`, or the error that `fs.ts` is to
/// call back with before it returns.
#[cold]
#[inline(never)]
fn reject_before_schedule(
global: &JSGlobalObject,
callback: Option<JSValue>,
error: JSValue,
delivery: EarlyDelivery,
) -> JsResult<JSValue> {
match callback {
None => Ok(JSPromise::rejected_promise(global, error).to_js()),
Some(callback) => {
let Some(callback) = callback else {
return Ok(JSPromise::rejected_promise(global, error).to_js());
};
match delivery {
// Returned, not called: `callback` is the `guardCallback` wrapper, and what the program's own function
// throws reaches the caller.
EarlyDelivery::BeforeReturn => Ok(error),
// The callback goes to nextTick bare: it asserts against an `AsyncContextFrame`.
EarlyDelivery::NextTick => {
JSValue::call_next_tick_1(callback, global, error)?;
Ok(JSValue::UNDEFINED)
}
Expand Down Expand Up @@ -174,7 +195,9 @@ impl Binding {
let arguments = JsLocal::to_js_slice(args);
let cp_args = match parse_async_args::<args::Cp<'static>>(global, arguments)? {
ParsedAsyncArgs::Args(args) => args,
ParsedAsyncArgs::Rejected(error) => return reject_before_schedule(global, None, error),
ParsedAsyncArgs::Rejected { error, delivery } => {
return reject_before_schedule(global, None, error, delivery);
}
};
let vm: &mut VirtualMachine = global.bun_vm().as_mut();
Ok(AsyncCpTask::create(&cx, self, cp_args, vm))
Expand Down Expand Up @@ -221,8 +244,8 @@ impl Binding {
let (callback, arguments) = split_callback(global, args, arm)?;
let rd_args = match parse_async_args::<args::Readdir<'static>>(global, arguments)? {
ParsedAsyncArgs::Args(args) => args,
ParsedAsyncArgs::Rejected(error) => {
return reject_before_schedule(global, callback, error);
ParsedAsyncArgs::Rejected { error, delivery } => {
return reject_before_schedule(global, callback, error, delivery);
}
};
let vm: &mut VirtualMachine = global.bun_vm().as_mut();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
// Native bindings of the library: readFileCb, writeFileCb and appendFileCb answer undefined, or the error of a signal that was aborted before the call, and then node:fs calls the program's callback before it returns. Code that takes the answer for undefined alone never calls it.
import { AsyncLocalStorage } from "node:async_hooks";
import { appendFile, mkdtempSync, readFile, readFileSync, rmSync, writeFile } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";

const folder = mkdtempSync(join(tmpdir(), "aot-fs-signal-"));
const file = join(folder, "file.txt");
const storage = new AsyncLocalStorage<string>();
const log: string[] = [];

function heard(name: string): (err: Error | null, data?: unknown) => void {
return (err, data) => {
log.push(`${name}: ${err === null ? "null" : err.name} ${typeof data} in ${storage.getStore()}`);
};
}

const aborted = { signal: AbortSignal.abort() };
const live = { signal: new AbortController().signal };

storage.run("the caller", () => {
writeFile(file, "never written", aborted, heard("writeFile, aborted"));
log.push("writeFile returned");
appendFile(file, "never written", aborted as any, heard("appendFile, aborted"));
log.push("appendFile returned");
readFile(file, aborted, heard("readFile, aborted"));
log.push("readFile returned");

try {
readFile(file, aborted, () => {
throw new Error("thrown by the callback");
});
log.push("nothing was thrown");
} catch (error) {
log.push(`the caller caught: ${(error as Error).message}`);
}

writeFile(file, "written", live, err => {
heard("writeFile")(err);
appendFile(file, " twice", err => {
heard("appendFile")(err);
readFile(file, "utf8", (err, data) => {
heard("readFile")(err, data);
log.push(`the file holds: ${readFileSync(file, "utf8")}`);
rmSync(folder, { recursive: true, force: true });
console.log(log.join("\n"));
});
});
});
log.push("writeFile with a live signal returned");
});
Loading
Loading