diff --git a/src/codegen/builtins/host/runtime-dependencies.builtins.ts b/src/codegen/builtins/host/runtime-dependencies.builtins.ts index af6be523ddb0..60c28ff0e6bf 100644 --- a/src/codegen/builtins/host/runtime-dependencies.builtins.ts +++ b/src/codegen/builtins/host/runtime-dependencies.builtins.ts @@ -65,6 +65,10 @@ const standardStreamModulesThatTestsCanWatch = standardStreamModules.map(load => ...load, when: b.whenTestsReplaceBuiltins, })); +// A standard output is a WriteStream, which writes through a FileSink and loads no node:fs. The input is a ReadStream, which reads through node:fs. +const standardOutputModulesThatTestsCanWatch = standardStreamModulesThatTestsCanWatch.filter( + load => load.module !== "node:fs", +); const loadedModulesByPath: Readonly> = { "node:fs/promises": evaluates("internal/fifo"), @@ -81,14 +85,14 @@ const loadedModulesByPath: Readonly ({ ...load, when: b.whenProgramStores("process.nextTick") })), ], - "process.stdout": standardStreamModulesThatTestsCanWatch, + "process.stdout": standardOutputModulesThatTestsCanWatch, }; const pluginFunctionsOfBuild = functionsIn( diff --git a/src/js/builtins/ProcessObjectInternals.ts b/src/js/builtins/ProcessObjectInternals.ts index 6589e3b43c4c..0a5b1695cf9d 100644 --- a/src/js/builtins/ProcessObjectInternals.ts +++ b/src/js/builtins/ProcessObjectInternals.ts @@ -71,8 +71,9 @@ export function getStdioWriteStream( }); stream._type = "tty"; } else { - const fs = require("node:fs"); - stream = new fs.WriteStream(null, { autoClose: false, fd, $fastPath: true }); + // The class itself, not fs.WriteStream: the stream does not load node:fs. + const { WriteStream } = require("internal/fs/streams"); + stream = new WriteStream(null, { autoClose: false, fd, $fastPath: true }); stream.readable = false; stream._type = "fs"; @@ -701,10 +702,8 @@ export function createOnWarning( // process.stderr.write, not console.error: under an inspector Session console.* emits // Runtime.consoleAPICalled, and a throwing listener there is surfaced via emitWarning, // so console.error would re-enter and loop. writeWarning reads process.stderr per call, as Node does. - // A compiled program may write without making the stream, whose modules put the operators of node:stream on Readable.prototype. - if (IS_COMPILED_AHEAD_OF_TIME && !redirectPath) require("internal/streams/load_all").rest?.(); const wroteInPlaceOfStream = writeWarning(message + "\n", redirectPath); - // Those modules also load node:events, which puts its EventEmitter.prototype above the prototype of `process`. Behind writeWarning(), which reads the members + // The modules of the stream load node:events, which puts its EventEmitter.prototype above the prototype of `process`. Behind writeWarning(), which reads the members // of built-in objects that they read, in their order. A program that cannot tell does without the module. if (IS_COMPILED_AHEAD_OF_TIME && WARNINGS_LOAD_NODE_EVENTS && wroteInPlaceOfStream) require("node:events"); } diff --git a/src/js/builtins/stdio.ts b/src/js/builtins/stdio.ts index a422b70ef45f..201ecc941c28 100644 --- a/src/js/builtins/stdio.ts +++ b/src/js/builtins/stdio.ts @@ -23,10 +23,7 @@ export function getLightStdioWriteStreams( // queued by the first read of it comes earlier than any other. And node:events puts its EventEmitter.prototype above the prototype of `process`. const nextTick = process.nextTick; require("node:events"); - // And all of node:stream puts its operators on Readable.prototype, which a program that has loaded a part of it reaches. - const streams = require("internal/streams/load_all"); - streams.rest?.(); - const served: ServedByLightStream = streams.servedByLightOutputs; + const served: ServedByLightStream = require("internal/streams/load_all").servedByLightOutputs; // What writeFast() of internal/fs/streams does with a receiver that is not a stream: it asks for the state of a stream and then calls _write(). function writeToForeignReceiver(receiver: any): never { if (receiver === undefined) @@ -163,9 +160,7 @@ export function getLightStdinStream(isTTY: boolean, needsWholeStream: NeedsWhole // As in getLightStdioWriteStreams(). void process.nextTick; require("node:events"); - const streams = require("internal/streams/load_all"); - streams.rest?.(); - const served: ServedByLightStream = streams.servedByLightInput; + const served: ServedByLightStream = require("internal/streams/load_all").servedByLightInput; return new Proxy( { get "%the whole stream%"() { diff --git a/src/js/internal/fs/streams.ts b/src/js/internal/fs/streams.ts index 793225c2bb73..dfd042176487 100644 --- a/src/js/internal/fs/streams.ts +++ b/src/js/internal/fs/streams.ts @@ -1,7 +1,10 @@ -// fs.ReadStream and fs.WriteStream are lazily loaded to avoid importing 'node:stream' until required -const { Readable, Writable, finished } = require("node:stream"); -const fs = require("node:fs"); -const { read, write, fsync, writev } = fs; +// process.stdout is a WriteStream: the Writable class and finished() load here, not node:stream, Readable or node:fs. +const { Stream } = require("internal/streams/legacy"); +// node:stream is `Stream` with its members: what a program has stored there, where node:stream is loaded. Else each loads alone. +const ReadableOfProgram: $Node.ReadableConstructor | undefined = $getByIdDirect(Stream, "Readable") ?? undefined; +const Writable: $Node.WritableConstructor = $getByIdDirect(Stream, "Writable") ?? require("internal/streams/writable"); +const finished: typeof Stream.finished = + $getByIdDirect(Stream, "finished") ?? require("internal/streams/end-of-stream"); const fileHandles = require("internal/fs/file_handles"); type FileHandleInternals = NonNullable; type FileHandle = $Node.FileHandle; @@ -73,7 +76,8 @@ declare global { flush?: boolean; open(): void; close(callback?: ErrorCallback): void; - [kFs]: StreamFs; + /** node:fs, or the `fs` option. Undefined on a WriteStream that writes through a FileSink. */ + [kFs]: StreamFs | undefined; [kIsPerformingIO]: boolean; /** true: the path must be opened; a sink: a FileSink. */ [kWriteStreamFastPath]?: undefined | true | FileSink; @@ -132,51 +136,56 @@ const { internals } = fileHandles; if (internals) takeFileHandleInternals(internals); else fileHandles.whenLoaded = takeFileHandleInternals; -const fileHandleStreamFs = (fh: FileHandle) => ({ - // try to use the basic fs.read/write/fsync if available, since they are less - // abstractions. however, node.js allows patching the file handle, so this has - // to be checked for. - read: - fh.read === fileHandlePrototypeRead - ? read - : function ( - fd: number | null, - buf: Buffer, - offset: number, - length: number, - pos: number | undefined, - cb: (error: Error | null, bytesRead?: number, buffer?: Buffer) => void, - ) { - return fh.read(buf, offset, length, pos).then( - ({ bytesRead, buffer }) => cb(null, bytesRead, buffer), - err => cb(err, 0, buf), - ); - }, - write: - fh.write === fileHandlePrototypeWrite - ? write - : function ( - fd: number | null, - buffer: Buffer, - offset: number, - length: number, - position: number | undefined, - cb: (error: Error | null, bytesWritten?: number, buffer?: Buffer) => void, - ) { - return fh.write(buffer, offset, length, position).then( - ({ bytesWritten, buffer }) => cb(null, bytesWritten, buffer), - err => cb(err, 0, buffer), - ); - }, - writev: fh.writev === fileHandlePrototypeWritev ? writev : undefined, - fsync: - fh.sync === fileHandlePrototypeFsync - ? fsync - : function (fd: number, cb: ErrorCallback) { - return fh.sync().then(() => cb(), cb); - }, - close: streamFileHandleClose.bind(fh), -}); +// node:fs loads with the first stream over a FileHandle. The stream calls these as node:fs writes them, not what its exports hold now. +function fileHandleStreamFs(fh: FileHandle) { + require("node:fs"); + const { read, write, fsync, writev } = require("internal/shared").originals.fs!; + return { + // try to use the basic fs.read/write/fsync if available, since they are less + // abstractions. however, node.js allows patching the file handle, so this has + // to be checked for. + read: + fh.read === fileHandlePrototypeRead + ? read + : function ( + fd: number | null, + buf: Buffer, + offset: number, + length: number, + pos: number | undefined, + cb: (error: Error | null, bytesRead?: number, buffer?: Buffer) => void, + ) { + return fh.read(buf, offset, length, pos).then( + ({ bytesRead, buffer }) => cb(null, bytesRead, buffer), + err => cb(err, 0, buf), + ); + }, + write: + fh.write === fileHandlePrototypeWrite + ? write + : function ( + fd: number | null, + buffer: Buffer, + offset: number, + length: number, + position: number | undefined, + cb: (error: Error | null, bytesWritten?: number, buffer?: Buffer) => void, + ) { + return fh.write(buffer, offset, length, position).then( + ({ bytesWritten, buffer }) => cb(null, bytesWritten, buffer), + err => cb(err, 0, buffer), + ); + }, + writev: fh.writev === fileHandlePrototypeWritev ? writev : undefined, + fsync: + fh.sync === fileHandlePrototypeFsync + ? fsync + : function (fd: number, cb: ErrorCallback) { + return fh.sync().then(() => cb(), cb); + }, + close: streamFileHandleClose.bind(fh), + }; +} function streamFileHandleClose(this: FileHandle, fd: FD, cb: ErrorCallback) { $assert(this[kFd] == fd, "fd mismatch"); @@ -270,9 +279,9 @@ function ReadStream( if (windowsHandle != null) { validateWindowsHandle(windowsHandle, fd, customFs); - this[kFs] = fs; + this[kFs] = require("node:fs"); } else if (fd == null) { - this[kFs] = customFs || fs; + this[kFs] = customFs || require("node:fs"); this.fd = null; this.path = getValidatedPath(path); const { flags, mode } = options; @@ -292,7 +301,7 @@ function ReadStream( validateInt32(fd, "fd", 0, 2147483647); } this.fd = fd as number; - this[kFs] = customFs || fs; + this[kFs] = customFs || require("node:fs"); } else if (typeof fd === "object" && FileHandle !== undefined && fd instanceof FileHandle) { if (options.fs) { throw $ERR_METHOD_NOT_IMPLEMENTED("FileHandle with fs"); @@ -340,12 +349,30 @@ function ReadStream( // such as highWaterMark and signal, and a throw there leaks the fd, as in Node. if (windowsHandle != null) this.fd = windowsHandleToFd(windowsHandle); - Readable.$call(this, options); + (ReadableOfProgram ?? require("internal/streams/readable")).$call(this, options); return this as unknown as void; } -$toClass(ReadStream, "ReadStream", Readable); +// Where node:stream is loaded, so is Readable: the class goes below what lies there now, and a stub there fails this load, which then runs again. +$toClass(ReadStream, "ReadStream", ReadableOfProgram); const readStreamPrototype = ReadStream.prototype; +// Else the class goes below the library's own Readable at its first use (`new`, `instanceof`, a read of `prototype`), so a WriteStream loads no Readable. +if (ReadableOfProgram === undefined) + Object.defineProperty(ReadStream, "prototype", { + get() { + const Readable = require("internal/streams/readable"); + Object.setPrototypeOf(readStreamPrototype, Readable.prototype); + Object.setPrototypeOf(ReadStream, Readable); + Object.defineProperty(ReadStream, "prototype", { value: readStreamPrototype, writable: true }); + return readStreamPrototype; + }, + set(value) { + Object.defineProperty(ReadStream, "prototype", { value, writable: true }); + }, + enumerable: false, + configurable: true, + }); + Object.defineProperty(readStreamPrototype, "autoClose", { get(this: $Node.ReadStream) { return this._readableState.autoDestroy; @@ -389,7 +416,7 @@ function streamConstruct(this: FileStream, callback: ErrorCallback) { return; } - this[kFs].open!(this.path, this.flags, this.mode, (err, fd) => { + this[kFs]!.open!(this.path, this.flags, this.mode, (err, fd) => { if (err) { callback(err); } else { @@ -417,7 +444,7 @@ readStreamPrototype._read = function (this: $Node.ReadStream, n: number) { const buf = Buffer.allocUnsafeSlow(n); this[kIsPerformingIO] = true; - this[kFs].read!(this.fd, buf, 0, n, this.pos, (er, bytesRead, buf) => { + this[kFs]!.read!(this.fd, buf, 0, n, this.pos, (er, bytesRead, buf) => { this[kIsPerformingIO] = false; // Tell ._destroy() that it's safe to close the fd now. @@ -491,7 +518,7 @@ function close(stream: FileStream, err: Error | null | undefined, cb: ErrorCallb if (stream.fd == null) { cb(err); } else if (stream.flush) { - stream[kFs].fsync!(stream.fd, flushErr => { + stream[kFs]!.fsync!(stream.fd, flushErr => { closeAfterSync(stream, err || flushErr, cb); }); } else { @@ -500,7 +527,7 @@ function close(stream: FileStream, err: Error | null | undefined, cb: ErrorCallb } function closeAfterSync(stream: FileStream, err: Error | null | undefined, cb: ErrorCallback) { - stream[kFs].close!(stream.fd!, er => { + stream[kFs]!.close!(stream.fd!, er => { cb(er || err); }); stream.fd = null; @@ -523,11 +550,13 @@ function WriteStream( options.decodeStrings = true; let { fd, autoClose, fs: customFs, start, flush, windowsHandle } = options; + // A stream that writes through a FileSink never reads node:fs, and process.stdout does not load it. + const fsOf = () => customFs || (fastPath ? undefined : require("node:fs")); if (windowsHandle != null) { validateWindowsHandle(windowsHandle, fd, customFs); - this[kFs] = fs; + this[kFs] = fsOf(); } else if (fd == null) { - this[kFs] = customFs || fs; + this[kFs] = fsOf(); this.fd = null; // Internal $fastPath callers (writableFromFileSink) discard .path; do not // resolve it - path.resolve("") needs process.cwd(), which throws when @@ -550,7 +579,7 @@ function WriteStream( validateInt32(fd, "fd", 0, 2147483647); } this.fd = fd as number; - this[kFs] = customFs || fs; + this[kFs] = fsOf(); } else if (typeof fd === "object" && FileHandle !== undefined && fd instanceof FileHandle) { if (options.fs) { throw $ERR_METHOD_NOT_IMPLEMENTED("FileHandle with fs"); @@ -579,7 +608,6 @@ function WriteStream( if (!writev) this._writev = null!; } else { this._writev = undefined; - $assert(this[kFs].write, "assuming user does not delete fs.write!"); } if (flush == null) { @@ -609,8 +637,10 @@ function WriteStream( fastWriter = Bun.file(fd as number).writer() as FileSink; } catch { fastPath = false; + this[kFs] = fsOf(); } } + $assert(fastPath || customFs || this[kFs]!.write, "assuming user does not delete fs.write!"); // Enable fast path if (fastPath) { @@ -661,7 +691,7 @@ function writeAll( cb: ErrorCallback, retries = 0, ) { - (IS_COMPILED_AHEAD_OF_TIME ? fileSystemOrThrow(this[kFs]) : this[kFs]).write!( + (IS_COMPILED_AHEAD_OF_TIME ? fileSystemOrThrow(this[kFs]) : this[kFs]!).write!( this.fd, data, 0, @@ -705,7 +735,7 @@ function writevAll( cb: ErrorCallback, retries = 0, ) { - this[kFs].writev!(this.fd, chunks, this.pos, (er, bytesWritten, buffers) => { + this[kFs]!.writev!(this.fd, chunks, this.pos, (er, bytesWritten, buffers) => { // No data currently available and operation should be retried later. if (er?.code === "EAGAIN") { er = null; diff --git a/src/js/internal/shared.ts b/src/js/internal/shared.ts index fcdd8a6e81f9..7b88a57515f4 100644 --- a/src/js/internal/shared.ts +++ b/src/js/internal/shared.ts @@ -461,7 +461,9 @@ const originals: { "emit" | "addListener" | "once" | "removeListener" | "removeAllListeners" | "listenerCount" > & { storeEmitWithRejectionCapture(emitter: $Node.EventEmitter): void }) | undefined; -} = { eventEmitter: undefined }; + // What internal/fs/streams calls for a stream over a FileHandle. + fs: Pick<$BunModules["node:fs"], "read" | "write" | "fsync" | "writev"> | undefined; +} = { eventEmitter: undefined, fs: undefined }; export default { isStoppedModuleGraphRunning, diff --git a/src/js/internal/stream.ts b/src/js/internal/stream.ts index 7256bf2ad1db..20e2b80771aa 100644 --- a/src/js/internal/stream.ts +++ b/src/js/internal/stream.ts @@ -1,10 +1,9 @@ "use strict"; -const ObjectKeys = Object.keys; -const ObjectDefineProperty = Object.defineProperty; +// Object lists a member from when it is first read, and loading this module has always read these two in this order. +const { keys: _ObjectKeys, defineProperty: ObjectDefineProperty } = Object; const customPromisify = Symbol.for("nodejs.util.promisify.custom"); -const { streamReturningOperators, promiseReturningOperators } = require("internal/streams/operators"); const compose = require("internal/streams/compose"); const { setDefaultHighWaterMark, getDefaultHighWaterMark } = require("internal/streams/state"); const { pipeline } = require("internal/streams/pipeline"); @@ -21,46 +20,6 @@ Stream.isReadable = utils.isReadable; Stream.isWritable = utils.isWritable; Stream.Readable = require("internal/streams/readable"); -const streamKeys = ObjectKeys(streamReturningOperators); -for (let i = 0; i < streamKeys.length; i++) { - const key = streamKeys[i]; - const op: (this: $Node.Readable, ...args: any[]) => AsyncIterable = streamReturningOperators[key]; - const fn = function (this: $Node.Readable, ...args: unknown[]) { - if (new.target) { - throw $ERR_ILLEGAL_CONSTRUCTOR(); - } - return Stream.Readable.from(op.$apply(this, args)); - }; - ObjectDefineProperty(fn, "name", { __proto__: null, value: op.name }); - ObjectDefineProperty(fn, "length", { __proto__: null, value: op.length }); - ObjectDefineProperty(Stream.Readable.prototype, key, { - __proto__: null, - value: fn, - enumerable: false, - configurable: true, - writable: true, - }); -} -const promiseKeys = ObjectKeys(promiseReturningOperators); -for (let i = 0; i < promiseKeys.length; i++) { - const key = promiseKeys[i]; - const op: (this: $Node.Readable, ...args: any[]) => Promise = promiseReturningOperators[key]; - const fn = function (this: $Node.Readable, ...args: unknown[]) { - if (new.target) { - throw $ERR_ILLEGAL_CONSTRUCTOR(); - } - return Promise.$resolve().then(() => op.$apply(this, args)); - }; - ObjectDefineProperty(fn, "name", { __proto__: null, value: op.name }); - ObjectDefineProperty(fn, "length", { __proto__: null, value: op.length }); - ObjectDefineProperty(Stream.Readable.prototype, key, { - __proto__: null, - value: fn, - enumerable: false, - configurable: true, - writable: true, - }); -} Stream.Writable = require("internal/streams/writable"); Stream.Duplex = require("internal/streams/duplex"); Stream.Transform = require("internal/streams/transform"); diff --git a/src/js/internal/streams/load_all.ts b/src/js/internal/streams/load_all.ts index 2935bbc1de64..b35d81f19a06 100644 --- a/src/js/internal/streams/load_all.ts +++ b/src/js/internal/streams/load_all.ts @@ -1,8 +1,5 @@ -// Making process.stdout or process.stderr loads all of node:stream, which adds members to what its parts define. The light streams that a compiled program may have -// in their place (builtins/stdio.ts) load none of it: where a part is loaded already, they ask here for the rest. +// What the light streams that a compiled program may have in place of process.stdout, process.stderr and process.stdin (builtins/stdio.ts) serve. export default { - /** Stored by the first part that is loaded. */ - rest: undefined as undefined | (() => void), // WHAT A LIGHT STREAM SERVES BY ITSELF. Its `get` trap asks these lists first, and the compiler keeps the whole stream out of a program that asks a stream for nothing else: // src/codegen/builtins/host/runtime-dependencies.builtins.ts reads them. A name that stands here and that no line of the trap answers, or the other way round, stops the program that asks for it. // What a method's name answers looks at its receiver, so the compiler lets it be called on the stream and do nothing else. diff --git a/src/js/internal/streams/operators.ts b/src/js/internal/streams/operators.ts index 1032b8935ba5..87f8f539c854 100644 --- a/src/js/internal/streams/operators.ts +++ b/src/js/internal/streams/operators.ts @@ -4,13 +4,7 @@ const { validateAbortSignal, validateFunction, validateInteger, validateObject } const { kWeakHandler, resistStopPropagation } = require("internal/shared"); const { finished } = require("internal/streams/end-of-stream"); -const MathFloor = Math.floor; -const PromiseResolve = Promise.$resolve.bind(Promise); -const PromiseReject = Promise.$reject.bind(Promise); const PromisePrototypeThen = $Promise.prototype.$then; -const ArrayPrototypePush = Array.prototype.push; -const NumberIsNaN = Number.isNaN; -const ObjectDefineProperty = Object.defineProperty; const kEmpty = Symbol("kEmpty"); const kEof = Symbol("kEof"); @@ -18,6 +12,14 @@ const kEof = Symbol("kEof"); type Operator = $Node.OperatorCallback; type Options = $Node.OperatorOptions | null | undefined; +// Math.floor without a read of Math, which user code can replace before this module loads. +function floor(value: number): number { + const number = +value; + const fraction = number % 1; + if (fraction !== fraction || fraction === 0) return number; + return fraction < 0 ? number - fraction - 1 : number - fraction; +} + function map(this: $Node.Readable, fn: Operator, options?: Options) { validateFunction(fn, "fn"); if (options != null) { @@ -29,12 +31,12 @@ function map(this: $Node.Readable, fn: Operator, options?: Options) { let concurrency = 1; if (options?.concurrency != null) { - concurrency = MathFloor(options.concurrency); + concurrency = floor(options.concurrency); } let highWaterMark = concurrency - 1; if (options?.highWaterMark != null) { - highWaterMark = MathFloor(options.highWaterMark); + highWaterMark = floor(options.highWaterMark); } validateInteger(concurrency, "options.concurrency", 1); @@ -88,9 +90,9 @@ function map(this: $Node.Readable, fn: Operator, options?: Options) { continue; } - val = PromiseResolve(val); + val = Promise.$resolve(val); } catch (err) { - val = PromiseReject(err); + val = Promise.$reject(err); } cnt += 1; @@ -111,7 +113,7 @@ function map(this: $Node.Readable, fn: Operator, options?: Options) { } queue.push(kEof); } catch (err) { - const val = PromiseReject(err); + const val = Promise.$reject(err); PromisePrototypeThen.$call(val, afterItemProcessed, onCatch); queue.push(val); } finally { @@ -280,7 +282,7 @@ async function toArray(this: $Node.Readable, options?: Options) { if (options?.signal?.aborted) { throw $makeAbortError(undefined, { cause: options.signal.reason }); } - ArrayPrototypePush.$call(result, val); + $arrayPush(result, val); } return result; } @@ -299,7 +301,7 @@ function toIntegerOrInfinity(number: number) { // We coerce here to align with the spec // https://github.com/tc39/proposal-iterator-helpers/issues/169 number = Number(number); - if (NumberIsNaN(number)) { + if (number !== number) { return 0; } if (number < 0) { @@ -331,7 +333,6 @@ function drop(this: $Node.Readable, number: number, options?: Options) { } }.$call(this); } -ObjectDefineProperty(drop, "length", { value: 1 }); function take(this: $Node.Readable, number: number, options?: Options) { if (options != null) { @@ -361,7 +362,6 @@ function take(this: $Node.Readable, number: number, options?: Options) { } }.$call(this); } -ObjectDefineProperty(take, "length", { value: 1 }); export default { streamReturningOperators: { diff --git a/src/js/internal/streams/readable.ts b/src/js/internal/streams/readable.ts index d0e89124c38b..ac30d9d5af1e 100644 --- a/src/js/internal/streams/readable.ts +++ b/src/js/internal/streams/readable.ts @@ -29,6 +29,7 @@ const isStandardOutputStream: (value: unknown) => boolean | undefined = $newCppF ); const ObjectDefineProperties = Object.defineProperties; +const ObjectDefineProperty = Object.defineProperty; const SymbolAsyncDispose = Symbol.asyncDispose; const NumberIsNaN = Number.isNaN; const NumberIsInteger = Number.isInteger; @@ -943,12 +944,7 @@ Readable.prototype.pipe = function (this: $Node.Readable, destination: $Node.Str $debug("pipe count=%d opts=%j", state.pipes.length, pipeOpts); let doEnd = !pipeOpts || pipeOpts.end !== false; - if (doEnd) { - const isStandardOutput = isStandardOutputStream(dest); - // Making process.stdout or process.stderr would have loaded all of node:stream. - if (isStandardOutput === undefined) require("internal/stream"); - doEnd = !isStandardOutput; - } + if (doEnd) doEnd = !isStandardOutputStream(dest); const endFn = doEnd ? onend : unpipe; if ((state[kState] & Flag.EndEmitted) !== 0) process.nextTick(endFn); @@ -1993,6 +1989,57 @@ Readable.wrap = function (src, options?) { }).wrap(src); }; -if (IS_COMPILED_AHEAD_OF_TIME) require("internal/streams/load_all").rest = () => void require("internal/stream"); +// Node defines the operators below in lib/stream.js. Here a user of this class may never load node:stream. +type Operators = typeof import("internal/streams/operators").default; +// Function.length of each operator. internal/streams/operators loads when one is first called. +type Lengths = Record & { __proto__: null }; +const streamOperatorLengths = { + __proto__: null, + drop: 1, + filter: 2, + flatMap: 2, + map: 2, + take: 1, +} satisfies Lengths<"streamReturningOperators">; +const promiseOperatorLengths = { + __proto__: null, + every: 1, + forEach: 2, + reduce: 3, + toArray: 1, + some: 1, + find: 2, +} satisfies Lengths<"promiseReturningOperators">; +type Operator = (this: $Node.Readable, ...args: any[]) => any; +let operators: Operators | undefined; + +function installOperator(key: string, length: number, returnsPromise: boolean) { + let op: Operator | undefined; + const fn = function (this: $Node.Readable, ...args: unknown[]) { + if (new.target) { + throw $ERR_ILLEGAL_CONSTRUCTOR(); + } + if (returnsPromise) { + return Promise.$resolve().$then(() => { + op ??= (operators ??= require("internal/streams/operators")).promiseReturningOperators[key] as Operator; + return op.$apply(this, args); + }); + } + op ??= (operators ??= require("internal/streams/operators")).streamReturningOperators[key] as Operator; + // node:stream is `Stream` with its members: what a program has stored under `Readable` there, once node:stream is loaded. + return ($getByIdDirect(Stream, "Readable") ?? Readable).from(op.$apply(this, args)); + }; + ObjectDefineProperty(fn, "name", { __proto__: null, value: key }); + ObjectDefineProperty(fn, "length", { __proto__: null, value: length }); + ObjectDefineProperty(Readable.prototype, key, { + __proto__: null, + value: fn, + enumerable: false, + configurable: true, + writable: true, + }); +} +for (const key in streamOperatorLengths) installOperator(key, streamOperatorLengths[key], false); +for (const key in promiseOperatorLengths) installOperator(key, promiseOperatorLengths[key], true); export default Readable as $Node.ReadableConstructor; diff --git a/src/js/node/fs.ts b/src/js/node/fs.ts index 3bcebaf18220..981723738767 100644 --- a/src/js/node/fs.ts +++ b/src/js/node/fs.ts @@ -24,7 +24,7 @@ const fs = require("internal/fs/binding"); const constants = $processBindingConstants.fs; -const { guardCallback, kCustomPromisifyArgsSymbol } = require("internal/shared"); +const { guardCallback, kCustomPromisifyArgsSymbol, originals } = 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; @@ -680,6 +680,9 @@ defineCustomPromisifyArgs(readv, ["bytesRead", "buffers"]); defineCustomPromisifyArgs(write, ["bytesWritten", "buffer"]); defineCustomPromisifyArgs(writev, ["bytesWritten", "buffers"]); +// The functions as they are written here: a stream over a FileHandle calls these, whatever a program assigns to the exports. +originals.fs = { read, write, fsync, writev }; + // The implementation (StatWatcher and friends) is lazily loaded from "internal/fs/watchfile" // the first time fs.watchFile or fs.unwatchFile is called. function watchFile(filename, options, listener?) { @@ -1298,7 +1301,7 @@ function globSync(pattern: string | string[], options?) { return Array.from(require("internal/fs/glob").globSync(pattern, options ?? kEmptyObject)); } -// The stream classes are accessors on `exports` so that loading node:fs does not load node:stream. The first read, or +// The stream classes are accessors on `exports` so that loading node:fs does not load the stream classes. The first read, or // an assignment, replaces the accessor with a data property. Once user code has sealed or frozen `exports` the accessor // is not configurable: Reflect.defineProperty returns false instead of throwing and the accessor stays in place. function materializeStream(name: string, value: T): T { @@ -1306,6 +1309,13 @@ function materializeStream(name: string, value: T): T { return value; } +// The read of `prototype` puts the class below Readable (internal/fs/streams defers that for process.stdout). +function readStream() { + const { ReadStream } = require("internal/fs/streams"); + void ReadStream.prototype; + return ReadStream; +} + var exports = { appendFile, appendFileSync, @@ -1411,7 +1421,7 @@ var exports = { Dir, Stats, get ReadStream() { - return materializeStream("ReadStream", require("internal/fs/streams").ReadStream); + return materializeStream("ReadStream", readStream()); }, set ReadStream(value) { materializeStream("ReadStream", value); @@ -1423,7 +1433,7 @@ var exports = { materializeStream("WriteStream", value); }, get FileReadStream() { - return materializeStream("FileReadStream", require("internal/fs/streams").ReadStream); + return materializeStream("FileReadStream", readStream()); }, set FileReadStream(value) { materializeStream("FileReadStream", value); diff --git a/src/js/node/http2.ts b/src/js/node/http2.ts index ea0a400f6fd1..7f59ffa44359 100644 --- a/src/js/node/http2.ts +++ b/src/js/node/http2.ts @@ -48,7 +48,7 @@ const kSessions = Symbol("sessions"); const kOptions = Symbol("options"); const kQuotedString = /^[\x09\x20-\x5b\x5d-\x7e\x80-\xff]*$/; const MAX_ADDITIONAL_SETTINGS = 10; -const Stream = require("node:stream"); +const Stream = require("internal/streams/legacy").Stream; const dc = require("node:diagnostics_channel"); // Built-in HTTP/2 diagnostics channels (mirror node's lib/internal/http2/core.js). @@ -64,7 +64,8 @@ const onServerStreamStartChannel = dc.channel("http2.server.stream.start"); const onServerStreamErrorChannel = dc.channel("http2.server.stream.error"); const onServerStreamFinishChannel = dc.channel("http2.server.stream.finish"); const onServerStreamCloseChannel = dc.channel("http2.server.stream.close"); -const { Readable } = Stream; +// node:stream is `Stream` with its members: what a program has stored there, where node:stream is loaded. Else the classes load alone. +const Readable: $Node.ReadableConstructor = $getByIdDirect(Stream, "Readable") ?? require("internal/streams/readable"); type Http2ConnectOptions = { settings?: Settings; protocol?: "https:" | "http:"; @@ -83,7 +84,8 @@ type Http2ConnectOptions = { }; const TLSSocket = tls.TLSSocket; const EventEmitter = require("node:events"); -const { Duplex } = Stream; +const Duplex: $Node.DuplexConstructor = $getByIdDirect(Stream, "Duplex") ?? require("internal/streams/duplex"); +const Writable = require("internal/streams/writable"); const { SafeArrayIterator, SafeSet } = require("internal/primordials"); const { promisify } = require("internal/promisify"); @@ -3256,7 +3258,7 @@ function doSendFileFD( end: typeof statOptions.length === "number" ? statOptions.length + (statOptions.offset || 0) - 1 : undefined, emitClose: false, }); - const sink = new Stream.Writable({ + const sink = new ($getByIdDirect(Stream, "Writable") ?? Writable)({ write(chunk, _encoding, cb) { const native = stream[bunHTTP2Session]?.[bunHTTP2Native]; if (!native || stream.destroyed || stream.closed) { diff --git a/src/js/node/zlib.ts b/src/js/node/zlib.ts index 675e7fa5e68b..d9f3805a375c 100644 --- a/src/js/node/zlib.ts +++ b/src/js/node/zlib.ts @@ -22,7 +22,12 @@ const MathMax = Math.max; const { isAnyArrayBuffer, isArrayBufferView } = require("node:util/types"); const kMaxLength = $requireMap.$get("buffer")?.exports.kMaxLength ?? BufferModule.kMaxLength; -const { Transform, finished } = require("node:stream"); +const { Stream } = require("internal/streams/legacy"); +// node:stream is `Stream` with its members: what a program has stored there, where node:stream is loaded. Else the two load alone. +const Transform: $Node.TransformConstructor = + $getByIdDirect(Stream, "Transform") ?? require("internal/streams/transform"); +const finished: typeof Stream.finished = + $getByIdDirect(Stream, "finished") ?? require("internal/streams/end-of-stream"); const owner_symbol = Symbol("owner_symbol"); const { checkRangesOrGetDefault, validateFunction, validateFiniteNumber } = require("internal/validators"); diff --git a/src/jsc/bindings/BunProcess.cpp b/src/jsc/bindings/BunProcess.cpp index e167cba8f19c..87dbbf5b5b3f 100644 --- a/src/jsc/bindings/BunProcess.cpp +++ b/src/jsc/bindings/BunProcess.cpp @@ -1869,10 +1869,9 @@ static void readMembersThatMakingStandardErrorReads(Zig::GlobalObject* globalObj { auto& vm = JSC::getVM(globalObject); auto scope = DECLARE_THROW_SCOPE(vm); - static constexpr ASCIILiteral membersOfObject[] = { "freeze"_s, "create"_s, "defineProperty"_s, "keys"_s, "setPrototypeOf"_s, "getPrototypeOf"_s, "defineProperties"_s, "assign"_s, "getOwnPropertyDescriptor"_s, "is"_s }; - static constexpr ASCIILiteral membersOfReflect[] = { "getPrototypeOf"_s, "ownKeys"_s, "getOwnPropertyDescriptor"_s, "defineProperty"_s }; + static constexpr ASCIILiteral membersOfObject[] = { "freeze"_s, "create"_s, "defineProperty"_s, "getPrototypeOf"_s, "defineProperties"_s, "assign"_s, "setPrototypeOf"_s, "is"_s }; + static constexpr ASCIILiteral membersOfReflect[] = { "ownKeys"_s, "getPrototypeOf"_s, "getOwnPropertyDescriptor"_s, "defineProperty"_s }; static constexpr ASCIILiteral membersOfSymbol[] = { "for"_s }; - static constexpr ASCIILiteral membersOfNumber[] = { "isNaN"_s }; static constexpr ASCIILiteral membersOfPromisePrototype[] = { "finally"_s }; readMembersOfGlobal(globalObject, "Object"_s, membersOfObject); RETURN_IF_EXCEPTION(scope, void()); @@ -1880,8 +1879,6 @@ static void readMembersThatMakingStandardErrorReads(Zig::GlobalObject* globalObj RETURN_IF_EXCEPTION(scope, void()); readMembersOfGlobal(globalObject, "Symbol"_s, membersOfSymbol); RETURN_IF_EXCEPTION(scope, void()); - readMembersOfGlobal(globalObject, "Number"_s, membersOfNumber); - RETURN_IF_EXCEPTION(scope, void()); JSValue promise = globalObject->get(globalObject, Identifier::fromString(vm, "Promise"_s)); RETURN_IF_EXCEPTION(scope, void()); if (!promise.isObject()) diff --git a/test/aot/a-built-in-module-that-threw-while-it-loaded-runs-again.test.ts b/test/aot/a-built-in-module-that-threw-while-it-loaded-runs-again.test.ts index c5403bd4ad29..6d6259a44ea5 100644 --- a/test/aot/a-built-in-module-that-threw-while-it-loaded-runs-again.test.ts +++ b/test/aot/a-built-in-module-that-threw-while-it-loaded-runs-again.test.ts @@ -36,7 +36,7 @@ runsAsPlain( ` const real = Object.keys; let first, second; - Object.keys = () => ["nonsense"]; + Object.keys = () => undefined; try { first = typeof require("node:zlib").gzipSync; } catch (error) { first = error.name; } finally { Object.keys = real; } try { second = typeof require("node:zlib").gzipSync; } catch (error) { second = error.name; } console.log(first, second);`, @@ -128,7 +128,7 @@ for (const [title, specifier] of [ return error.name; } }; - Object.keys = () => ["nonsense"]; + Object.keys = () => undefined; const first = await imported(); Object.keys = real; console.log(first, await imported(), load("node:zlib"), await imported());`, diff --git a/test/aot/parallel/library-modules/stream-operators-arrive-with-first-pipe.js b/test/aot/parallel/library-modules/stream-operators-come-with-the-readable-class.js similarity index 89% rename from test/aot/parallel/library-modules/stream-operators-arrive-with-first-pipe.js rename to test/aot/parallel/library-modules/stream-operators-come-with-the-readable-class.js index 3515bdcdfe3d..5f54478e5239 100644 --- a/test/aot/parallel/library-modules/stream-operators-arrive-with-first-pipe.js +++ b/test/aot/parallel/library-modules/stream-operators-come-with-the-readable-class.js @@ -1,5 +1,5 @@ -// With node:net alone, Readable.prototype has no map(), filter() and so on. They arrive with all of node:stream, which the first pipe() that looks at -// process.stdout loads. Each way is a process of its own. +// With node:net alone, Readable.prototype has map(), filter() and so on: the module of the class defines them. They do not wait for node:stream, which +// the first pipe() that looks at process.stdout loads. Each way is a process of its own. const ways = { "nothing"() {}, "console.log"() { diff --git a/test/aot/the-first-read-of-a-standard-stream-calls-what-a-test-has-replaced.test.ts b/test/aot/the-first-read-of-a-standard-stream-calls-what-a-test-has-replaced.test.ts index ed6caf596283..f1281c952a65 100644 --- a/test/aot/the-first-read-of-a-standard-stream-calls-what-a-test-has-replaced.test.ts +++ b/test/aot/the-first-read-of-a-standard-stream-calls-what-a-test-has-replaced.test.ts @@ -2,26 +2,29 @@ import { expect } from "bun:test"; import { casesOfOneRun, sayOf } from "./harness/cases-of-one-run"; import { test } from "./supported"; -// No case asks a stream for more than `fd` and `write`. +// No case asks a stream for more than `fd` and `write`. The modules behind a standard stream call Object.getPrototypeOf when +// they load (node:events), and Reflect.ownKeys when they are used (node:events, internal/streams/legacy). A replaced +// Object.getPrototypeOf that answers nonsense makes the load throw. node:child_process loads node:events too: a case about +// it alone replaces Reflect.ownKeys, which its load does not call. const project = casesOfOneRun("the-first-read-of-a-standard-stream-calls-what-a-test-has-replaced", { "package.json": `{ "name": "p", "type": "module" }`, "tsconfig.json": "{}", }); -function runsAsPlain(title: string, expected: string, body: string) { +function runsAsPlain(title: string, expected: string, body: string, replaced = "Object.getPrototypeOf") { const name = project.add( name => ` import { spyOn, test } from "bun:test"; ${sayOf(name)} - const keys = Object.keys; - const nonsense = () => ["nonsense"]; + const real = ${replaced}; + const nonsense = () => undefined; test("${name}", () => { - try {${body.replaceAll("console.log(", "say(")} + try {${body.replaceAll("console.log(", "say(").replaceAll("REPLACED", replaced)} } catch (error) { - Object.keys = keys; + ${replaced} = real; say("threw", error.name); } finally { - Object.keys = keys; + ${replaced} = real; } });`, ); @@ -35,59 +38,59 @@ function runsAsPlain(title: string, expected: string, body: string) { for (const stream of ["stdout", "stderr"]) { runsAsPlain( - `process.${stream}.write() calls an Object.keys that counts`, + `process.${stream}.write() calls an Object.getPrototypeOf that counts`, "true true", ` let calls = 0; - Object.keys = object => (calls++, keys(object)); + REPLACED = object => (calls++, real(object)); const written = process.${stream}.write(""); - Object.keys = keys; + REPLACED = real; console.log(written, calls > 0);`, ); runsAsPlain( - `process.${stream} is undefined behind an Object.keys that its modules throw on`, + `process.${stream} is undefined behind an Object.getPrototypeOf that its modules throw on`, "undefined", ` - Object.keys = nonsense; + REPLACED = nonsense; const kind = typeof process.${stream}; - Object.keys = keys; + REPLACED = real; console.log(kind);`, ); runsAsPlain( - `process.${stream}.write() throws behind an Object.keys that its modules throw on`, + `process.${stream}.write() throws behind an Object.getPrototypeOf that its modules throw on`, "threw TypeError", ` - Object.keys = nonsense; + REPLACED = nonsense; console.log(process.${stream}.write(""));`, ); } runsAsPlain( - "process.stdin.fd calls an Object.keys that counts", + "process.stdin.fd calls an Object.getPrototypeOf that counts", "0 true", ` let calls = 0; - Object.keys = object => (calls++, keys(object)); + REPLACED = object => (calls++, real(object)); const fd = process.stdin.fd; - Object.keys = keys; + REPLACED = real; console.log(fd, calls > 0);`, ); runsAsPlain( - "process.stdin is undefined behind an Object.keys that its modules throw on", + "process.stdin is undefined behind an Object.getPrototypeOf that its modules throw on", "undefined", ` - Object.keys = nonsense; + REPLACED = nonsense; const kind = typeof process.stdin; - Object.keys = keys; + REPLACED = real; console.log(kind);`, ); runsAsPlain( - "a spy on Object.keys hears the first write", + "a spy on Object.getPrototypeOf hears the first write", "true", ` - const spy = spyOn(Object, "keys"); + const spy = spyOn(Object, "getPrototypeOf"); process.stdout.write(""); const heard = spy.mock.calls.length > 0; spy.mockRestore(); @@ -112,7 +115,7 @@ runsAsPlain( "execSync(), which hands on what the child wrote to its stderr", "threw TypeError", ` - Object.keys = nonsense; + REPLACED = nonsense; console.log(String(require("node:child_process").execSync("echo ex")));`, ); @@ -120,6 +123,7 @@ runsAsPlain( "execSync() with the streams of the child named reads no stream of its own", "ex", ` - Object.keys = nonsense; + REPLACED = nonsense; console.log(String(require("node:child_process").execSync("echo ex", { stdio: "pipe" })).trim());`, + "Reflect.ownKeys", ); diff --git a/test/internal/source-lints/stream-barrel-in-builtins.test.ts b/test/internal/source-lints/stream-barrel-in-builtins.test.ts new file mode 100644 index 000000000000..b7665b158360 --- /dev/null +++ b/test/internal/source-lints/stream-barrel-in-builtins.test.ts @@ -0,0 +1,43 @@ +import { Glob } from "bun"; +import { expect, test } from "bun:test"; +import { readFileSync } from "node:fs"; +import path from "node:path"; + +// `require("node:stream")` evaluates internal/stream.ts, the barrel: operators, compose, pipeline, Duplex, Transform, +// PassThrough, and everything those load. A built-in module that needs one stream class requires that class from +// internal/streams/* (the barrel re-exports the same objects, so `instanceof` holds), or requires the barrel inside +// the function that needs it. process.stdout, process.stderr, node:zlib and node:http once paid the whole barrel at +// load (oven-sh/bun#48005). +// +// A require at the top level of a module starts its line in column 0. One inside a function is indented, and runs on +// first use. Known: the modules that are node:stream, and node:http, whose classes extend what a program has laid on +// node:stream when it loads (test/aot/parallel/what-node-http-takes-off-node-stream). +test("built-in modules do not require the node:stream barrel at load", async () => { + const repoRoot = path.resolve(import.meta.dir, "..", "..", ".."); + const barrel = /^\S.*\brequire\("(?:node:stream|internal\/stream)"\)/; + + const known = new Map([ + ["src/js/ahead-of-time/internal/http/FakeSocket.ts", 1], + ["src/js/ahead-of-time/node/_http_outgoing.ts", 1], + ["src/js/ahead-of-time/node/_http_server.ts", 1], + ["src/js/internal/http/FakeSocket.ts", 1], + ["src/js/node/_http_outgoing.ts", 1], + ["src/js/node/_http_server.ts", 1], + ["src/js/node/_stream_wrap.ts", 1], + ["src/js/node/stream.promises.ts", 1], + ["src/js/node/stream.ts", 1], + ]); + + const found = new Map(); + let scanned = 0; + for await (const rel of new Glob("**/*.{ts,js}").scan({ cwd: path.join(repoRoot, "src/js") })) { + const file = path.join("src/js", rel).replaceAll("\\", "/"); + if (file.endsWith(".d.ts")) continue; + scanned++; + const lines = readFileSync(path.join(repoRoot, file), "utf8").split("\n"); + const count = lines.filter(line => !/^\/[/*]/.test(line) && barrel.test(line)).length; + if (count > 0) found.set(file, count); + } + expect(scanned).toBeGreaterThan(100); + expect(Object.fromEntries([...found].sort())).toEqual(Object.fromEntries([...known].sort())); +}); diff --git a/test/js/node/fs/fs-filehandle-stream-replaced-fs-fixture.cjs b/test/js/node/fs/fs-filehandle-stream-replaced-fs-fixture.cjs new file mode 100644 index 000000000000..7862cfa7eee3 --- /dev/null +++ b/test/js/node/fs/fs-filehandle-stream-replaced-fs-fixture.cjs @@ -0,0 +1,107 @@ +// A program puts its own read, write, writev and fsync on node:fs, as a file system held in memory does: each counts +// the call, reports success and touches no file. A stream over a FileHandle does its I/O through the handle +// (FileHandleOperations in Node's lib/internal/fs/streams.js), so it calls none of them and still reaches the file. +// Only a stream over a path or a raw descriptor calls what node:fs exports. Node.js prints the same for this file. +// +// argv[2]: when the program puts its functions there. The working directory holds lines.txt and nothing else. +"use strict"; +const when = process.argv[2]; + +// node:fs keeps the functions for these streams in an object of its own. A program's accessor on Object.prototype +// must not see that: node:fs loads, and the setter is handed nothing. +let setterCalls = 0; +if (when === "under an accessor named fs on Object.prototype") { + Object.defineProperty(Object.prototype, "fs", { + get() { + return undefined; + }, + set() { + setterCalls++; + }, + configurable: true, + }); +} +if (when === "after process.stdout") void process.stdout; +const fs = require("node:fs"); +const { once } = require("node:events"); +if (when === "after the stream classes") { + void fs.ReadStream; + void fs.WriteStream; +} + +const calls = { read: 0, write: 0, writev: 0, fsync: 0 }; +fs.read = (fd, buffer, offset, length, position, callback) => { + calls.read++; + process.nextTick(callback, null, 0, buffer); +}; +fs.write = (fd, buffer, offset, length, position, callback) => { + calls.write++; + process.nextTick(callback, null, length, buffer); +}; +fs.writev = (fd, buffers, position, callback) => { + calls.writev++; + process.nextTick(callback, null, Buffer.concat(buffers).length, buffers); +}; +fs.fsync = (fd, callback) => { + calls.fsync++; + process.nextTick(callback, null); +}; + +// What the stream leaves in the file. +async function written(file, stream, write = () => stream.end("abc")) { + write(); + await once(stream, "close"); + return fs.readFileSync(file, "utf8"); +} + +// What the stream reads. +async function read(stream) { + let text = ""; + stream.on("data", chunk => (text += chunk)); + await once(stream, "close"); + return text; +} + +(async () => { + const open = (file, flags) => fs.promises.open(file, flags); + const corked = (await open("3.txt", "w")).createWriteStream(); + const overHandle = { + "handle.createWriteStream()": await written("1.txt", (await open("1.txt", "w")).createWriteStream()), + "fs.createWriteStream(null, { fd })": await written( + "2.txt", + fs.createWriteStream(null, { fd: await open("2.txt", "w") }), + ), + "a corked batch": await written("3.txt", corked, () => { + corked.cork(); + corked.write("a"); + corked.write("b"); + corked.write("c"); + corked.uncork(); + corked.end(); + }), + "flush": await written("4.txt", (await open("4.txt", "w")).createWriteStream({ flush: true })), + "handle.createReadStream()": await read((await open("lines.txt")).createReadStream()), + "fs.createReadStream(null, { fd })": await read(fs.createReadStream(null, { fd: await open("lines.txt") })), + calls: { ...calls }, + }; + // The controls: a stream over a path or a raw descriptor calls the program's functions, which write nothing and + // read nothing. + let before = { ...calls }; + const overPath = { + "fs.createWriteStream(path)": await written("5.txt", fs.createWriteStream("5.txt")), + "fs.createReadStream(path)": await read(fs.createReadStream("lines.txt")), + calls: { read: calls.read - before.read, write: calls.write - before.write }, + }; + before = { ...calls }; + const overDescriptor = { + "fs.createWriteStream(null, { fd: number })": await written( + "6.txt", + fs.createWriteStream(null, { fd: fs.openSync("6.txt", "w") }), + ), + "fs.createReadStream(null, { fd: number })": await read( + fs.createReadStream(null, { fd: fs.openSync("lines.txt") }), + ), + calls: { read: calls.read - before.read, write: calls.write - before.write }, + }; + console.log(JSON.stringify({ overHandle, overPath, overDescriptor, setterCalls })); +})(); diff --git a/test/js/node/fs/fs.test.ts b/test/js/node/fs/fs.test.ts index 4107357a905d..e41a50f87bd0 100644 --- a/test/js/node/fs/fs.test.ts +++ b/test/js/node/fs/fs.test.ts @@ -437,6 +437,53 @@ describe("FileHandle", () => { expect(readFileSync(path, "utf8")).toBe("Test file written successfully"); }); + // A program puts its own read, write, writev and fsync on node:fs, as a file system held in memory does. Node's + // stream over a FileHandle does its I/O through the handle (FileHandleOperations in lib/internal/fs/streams.js): it + // calls none of them and reaches the file. A stream over a path or a raw descriptor calls them. The fixture prints + // the same under Node.js. + it.concurrent.each([ + "first", + "after the stream classes", + "after process.stdout", + "under an accessor named fs on Object.prototype", + ])("a stream over a FileHandle does not call what a program puts on node:fs %s", async when => { + using dir = tempDir("fs-filehandle-stream-replaced-fs", { "lines.txt": "a\nb\nc\n" }); + await using proc = Bun.spawn({ + cmd: [bunExe(), join(import.meta.dir, "fs-filehandle-stream-replaced-fs-fixture.cjs"), when], + env: bunEnv, + cwd: String(dir), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ result: JSON.parse(stdout), stderr }).toEqual({ + result: { + overHandle: { + "handle.createWriteStream()": "abc", + "fs.createWriteStream(null, { fd })": "abc", + "a corked batch": "abc", + "flush": "abc", + "handle.createReadStream()": "a\nb\nc\n", + "fs.createReadStream(null, { fd })": "a\nb\nc\n", + calls: { read: 0, write: 0, writev: 0, fsync: 0 }, + }, + overPath: { + "fs.createWriteStream(path)": "", + "fs.createReadStream(path)": "", + calls: { read: 1, write: 1 }, + }, + overDescriptor: { + "fs.createWriteStream(null, { fd: number })": "", + "fs.createReadStream(null, { fd: number })": "", + calls: { read: 1, write: 1 }, + }, + setterCalls: 0, + }, + stderr: "", + }); + expect(exitCode).toBe(0); + }); + // Node.js closes a FileHandle's fd in its native finalizer and raises // ERR_INVALID_STATE (DEP0137 end-of-life) when the handle is collected // without close(). Bun must reclaim the fd and surface the same diagnostic. diff --git a/test/js/node/process/process-stdio.test.ts b/test/js/node/process/process-stdio.test.ts index 8d02e2de3cab..14965eacd038 100644 --- a/test/js/node/process/process-stdio.test.ts +++ b/test/js/node/process/process-stdio.test.ts @@ -186,6 +186,175 @@ describe.concurrent("process-stdio", () => { }); }); +// The first read of process.stdout or process.stderr makes an fs.WriteStream. What that loads is the Writable class and what it +// needs: not node:stream as a whole (which adds pipeline, compose and the other classes to the Stream function that Writable +// inherits from), not Readable, and not node:fs. oven-sh/bun#48005: the barrel cost a first read of process.stdout about as +// much as an empty process. A built-in module that loads calls Object.defineProperty. A require of one that is loaded calls +// nothing: the count of calls under a require says whether the stream had loaded the module. +describe.concurrent("the first read of a standard stream loads only what it needs", () => { + // `bun -e` of a CommonJS text loads the built-in module that a top-level name of the text names: no binding here is + // called `stream` or `fs`. + const whatIsLoaded = (name: string) => ` + const standard = process.${name}; + let calls = 0; + const real = Object.defineProperty; + Object.defineProperty = function (...args) { calls++; return real.apply(this, args); }; + const nodeFs = require("node:fs"); + const fsLoads = calls > 0; + calls = 0; + require("node:_stream_readable"); + const readableLoads = calls > 0; + Object.defineProperty = real; + let Stream = standard.constructor; + while (Stream.name !== "Stream") Stream = Object.getPrototypeOf(Stream); + process.${name}.write(JSON.stringify({ + write: typeof standard.write, + barrel: typeof Stream.pipeline, + readable: "Readable" in Stream, + fsWriteStream: typeof Object.getOwnPropertyDescriptor(nodeFs, "WriteStream").get, + fsReadStream: typeof Object.getOwnPropertyDescriptor(nodeFs, "ReadStream").get, + fsLoads, + readableLoads, + }) + "\\n"); + `; + const writeStreamOnly = { + write: "function", + barrel: "undefined", + readable: false, + fsWriteStream: "function", + fsReadStream: "function", + fsLoads: true, + readableLoads: true, + }; + + test.each(["stdout", "stderr"] as const)("process.%s on a pipe", async name => { + await using proc = spawn({ + cmd: [bunExe(), "-e", whatIsLoaded(name)], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const [written, other] = name === "stdout" ? [stdout, stderr] : [stderr, stdout]; + expect({ loaded: JSON.parse(written), other }).toEqual({ loaded: writeStreamOnly, other: "" }); + expect(exitCode).toBe(0); + }); + + test.each(["stdout", "stderr"] as const)("process.%s on a file", async name => { + using dir = tempDir("stdio-loads", {}); + const file = Bun.file(path.join(String(dir), name)); + await using proc = spawn({ + cmd: [bunExe(), "-e", whatIsLoaded(name)], + env: bunEnv, + stdout: name === "stdout" ? file : "pipe", + stderr: name === "stderr" ? file : "pipe", + }); + const [other, exitCode] = await Promise.all([(name === "stdout" ? proc.stderr : proc.stdout).text(), proc.exited]); + expect({ loaded: JSON.parse(await file.text()), other }).toEqual({ loaded: writeStreamOnly, other: "" }); + expect(exitCode).toBe(0); + }); + + test("process.stdin is a Readable of node:fs, and still loads no node:stream", async () => { + const script = ` + const standard = process.stdin; + let Stream = standard.constructor; + while (Stream.name !== "Stream") Stream = Object.getPrototypeOf(Stream); + console.log(JSON.stringify({ + read: typeof standard.read, + map: typeof standard.map, + barrel: typeof Stream.pipeline, + })); + `; + await using proc = spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdin: new Blob(["x"]), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ loaded: JSON.parse(stdout), stderr }).toEqual({ + loaded: { read: "function", map: "function", barrel: "undefined" }, + stderr: "", + }); + expect(exitCode).toBe(0); + }); + + test("node:stream, loaded after process.stdout, is the class of process.stdout", async () => { + const script = ` + const standard = process.stdout; + const nodeFs = require("node:fs"); + const { Writable, Readable, Stream } = require("node:stream"); + console.log(JSON.stringify({ + writable: standard instanceof Writable, + stream: standard instanceof Stream, + fsWriteStream: standard instanceof nodeFs.WriteStream, + readStream: new nodeFs.ReadStream(process.execPath) instanceof Readable, + })); + `; + await using proc = 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({ loaded: JSON.parse(stdout), stderr }).toEqual({ + loaded: { writable: true, stream: true, fsWriteStream: true, readStream: true }, + stderr: "", + }); + expect(exitCode).toBe(0); + }); + + // fs.ReadStream goes below Readable at the first read of its `prototype`. What node:fs hands out has been there: its + // statics and its prototype chain are whole before anything is constructed. + test("fs.ReadStream is whole before the first instance", async () => { + const script = ` + process.stdout; + const nodeFs = require("node:fs"); + const describe = ReadStream => { + const { get, writable, enumerable } = Object.getOwnPropertyDescriptor(ReadStream, "prototype"); + return { get: typeof get, writable, enumerable, from: typeof ReadStream.from, parent: Object.getPrototypeOf(ReadStream).name }; + }; + console.log(JSON.stringify({ ReadStream: describe(nodeFs.ReadStream), FileReadStream: describe(nodeFs.FileReadStream) })); + `; + await using proc = 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]); + const whole = { get: "undefined", writable: true, enumerable: false, from: "function", parent: "Readable" }; + expect({ loaded: JSON.parse(stdout), stderr }).toEqual({ + loaded: { ReadStream: whole, FileReadStream: whole }, + stderr: "", + }); + expect(exitCode).toBe(0); + }); + + // process.stdout and process.stderr are built from the class in internal/fs/streams, not from node:fs, which they do + // not load. A replaced fs.WriteStream is not their class (Node's are a net.Socket or a SyncWriteStream, never an + // fs.WriteStream). process.stdin comes from node:fs: a replaced fs.ReadStream is its class, as in Node for a file. + test("a replaced fs.WriteStream is not the class of process.stdout, a replaced fs.ReadStream is the class of process.stdin", async () => { + const script = ` + const nodeFs = require("node:fs"); + class Out extends nodeFs.WriteStream {} + class In extends nodeFs.ReadStream {} + nodeFs.WriteStream = Out; + nodeFs.ReadStream = In; + console.log(JSON.stringify({ + stdout: process.stdout.constructor.name, + stderr: process.stderr.constructor.name, + stdin: process.stdin.constructor.name, + })); + `; + await using proc = spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdin: new Blob(["x"]), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ loaded: JSON.parse(stdout), stderr }).toEqual({ + loaded: { stdout: "WriteStream", stderr: "WriteStream", stdin: "In" }, + stderr: "", + }); + expect(exitCode).toBe(0); + }); +}); + // Like Node, process.stdout/stderr put O_NONBLOCK on fd 1/2 when they are pipes or sockets. The flag lives on the shared open // file description, so spawn clears it again for a child's fds 0-2 (libuv does the same), and the native console writer must // wait on EAGAIN instead of dropping the rest of the line. diff --git a/test/js/node/stream/node-stream-reads-and-messages.test.ts b/test/js/node/stream/node-stream-reads-and-messages.test.ts index a6640844d65b..5e18050956ae 100644 --- a/test/js/node/stream/node-stream-reads-and-messages.test.ts +++ b/test/js/node/stream/node-stream-reads-and-messages.test.ts @@ -1,6 +1,7 @@ // What a program can see of node:stream and node:events besides their answers: which properties of a stream they read and store, seen -// from an object below a Proxy and from accessors, what an error says when a member of a stream is not a function, and when the -// operators of Readable.prototype arrive. The expected files hold what each program printed before the modules were typed. +// from an object below a Proxy and from accessors, what an error says when a member of a stream is not a function, and that the +// operators of Readable.prototype are there whatever is loaded first. The first two expected files hold what each program printed +// before the modules were typed. import { expect, test } from "bun:test"; import { bunEnv, bunExe } from "harness"; import { join } from "node:path"; @@ -10,7 +11,7 @@ const programs = join(import.meta.dir, "../../../aot/parallel/library-modules"); test.concurrent.each([ ["stream-reads-seen-without-proxy-receiver", { STREAM_MEMBERS: "all", STREAM_READS: "all" }], ["stream-members-replaced-by-not-function", { MESSAGES: "whole" }], - ["stream-operators-arrive-with-first-pipe", {}], + ["stream-operators-come-with-the-readable-class", {}], ])( "%s", async (name, env) => { diff --git a/test/js/node/stream/node-stream.test.js b/test/js/node/stream/node-stream.test.js index b7ad335a3448..6df9207697f0 100644 --- a/test/js/node/stream/node-stream.test.js +++ b/test/js/node/stream/node-stream.test.js @@ -1954,6 +1954,329 @@ describe("stream operators argument validation (nodejs/node#59529)", () => { }); }); +// Node defines the operator methods when lib/stream.js loads, and its net, tls, crypto, child_process and +// worker_threads load that file. In Bun those modules, _http_incoming and the _stream_* modules reach the class +// and never load node:stream. Each script below runs in a fresh process that has not loaded node:stream. +// Neither does a read of process.stdout, process.stderr or process.stdin, nor console.log (process-stdio.test.ts pins that). +describe("Readable operators in a process that has not loaded node:stream", () => { + const lengths = { + drop: 1, + filter: 2, + flatMap: 2, + map: 2, + take: 1, + every: 1, + forEach: 2, + reduce: 3, + toArray: 1, + some: 1, + find: 2, + }; + const operators = Object.keys(lengths); + + async function run(script) { + // `bun -e` exposes the builtin modules as globals, and a top-level `const stream = ...` in the + // script would be enough to load node:stream. Block scoping the script keeps it away from them. + 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]); + return { stdout, stderr, exitCode }; + } + + // Each row: a title, a setup statement, and the streams to check as { name: expression }. + it.concurrent.each([ + ["node:net", ``, { socket: `new (require("node:net").Socket)()` }], + ["node:tls", ``, { socket: `new (require("node:tls").TLSSocket)()` }], + [ + "node:crypto", + `const crypto = require("node:crypto"), key = Buffer.alloc(16);`, + { + hash: `crypto.createHash("sha256")`, + hmac: `crypto.createHmac("sha256", key)`, + cipher: `crypto.createCipheriv("aes-128-cbc", key, key)`, + decipher: `crypto.createDecipheriv("aes-128-cbc", key, key)`, + }, + ], + [ + "node:child_process, stdin not a pipe", + `const child = require("node:child_process").spawn(process.execPath, ["-e", ""], { + stdio: ["ignore", "pipe", "pipe", "pipe"], + });`, + { stdout: `child.stdout`, stderr: `child.stderr`, "stdio[3]": `child.stdio[3]` }, + ], + [ + "node:worker_threads, the parent's end", + `const worker = new (require("node:worker_threads").Worker)("", { eval: true, stdout: true, stderr: true });`, + { stdout: `worker.stdout`, stderr: `worker.stderr` }, + ], + ["node:_http_incoming", ``, { prototype: `require("node:_http_incoming").IncomingMessage.prototype` }], + ["node:_stream_readable", ``, { prototype: `require("node:_stream_readable").prototype` }], + ["node:_stream_duplex", ``, { prototype: `require("node:_stream_duplex").prototype` }], + ["node:_stream_transform", ``, { prototype: `require("node:_stream_transform").prototype` }], + ["node:_stream_passthrough", ``, { prototype: `require("node:_stream_passthrough").prototype` }], + ])("%s has all eleven", async (_, setup, targets) => { + const names = Object.keys(targets); + const result = await run(` + ${setup} + const targets = { ${names.map(name => `${JSON.stringify(name)}: ${targets[name]}`).join(", ")} }; + const missing = {}; + for (const name in targets) { + missing[name] = ${JSON.stringify(operators)}.filter(operator => typeof targets[name][operator] !== "function"); + } + console.log(JSON.stringify(missing)); + `); + expect(result).toEqual({ + stdout: JSON.stringify(Object.fromEntries(names.map(name => [name, []]))) + "\n", + stderr: "", + exitCode: 0, + }); + }); + + it.concurrent("process.stdin inside a Worker has all eleven", async () => { + const inWorker = ` + const stdin = process.stdin; + const missing = ${JSON.stringify(operators)}.filter(operator => typeof stdin[operator] !== "function"); + require("node:worker_threads").parentPort.postMessage(missing); + `; + const result = await run(` + const worker = new (require("node:worker_threads").Worker)(${JSON.stringify(inWorker)}, { eval: true }); + worker.on("message", missing => console.log(JSON.stringify({ stdin: missing }))); + `); + expect(result).toEqual({ stdout: JSON.stringify({ stdin: [] }) + "\n", stderr: "", exitCode: 0 }); + }); + + it.concurrent("they are the last eleven own properties, with Node's names, lengths and attributes", async () => { + const result = await run(` + const { prototype } = require("node:_stream_readable"); + const found = Object.getOwnPropertyNames(prototype).slice(-11).map(key => { + const { value, writable, enumerable, configurable } = Object.getOwnPropertyDescriptor(prototype, key); + return { key, name: value.name, length: value.length, writable, enumerable, configurable }; + }); + const duplexHasOwnCopy = Object.hasOwn(require("node:_stream_duplex").prototype, "map"); + console.log(JSON.stringify({ found, duplexHasOwnCopy })); + `); + expect({ ...result, stdout: JSON.parse(result.stdout || "null") }).toEqual({ + stdout: { + found: operators.map(key => ({ + key, + name: key, + length: lengths[key], + writable: true, + enumerable: false, + configurable: true, + })), + duplexHasOwnCopy: false, + }, + stderr: "", + exitCode: 0, + }); + }); + + it.concurrent("all eleven work when the first call is what loads their implementation", async () => { + const result = await run(` + const Readable = require("node:_stream_readable"); + const source = () => Readable.from([1, 2, 3, 4]); + (async () => { + let forEach = 0; + await source().forEach(n => { + forEach += n; + }); + console.log( + JSON.stringify({ + drop: await source().drop(3).toArray(), + filter: await source().filter(n => n % 2 === 0).toArray(), + flatMap: await source().flatMap(n => [n, -n]).toArray(), + map: await source().map(n => n * 10).toArray(), + take: await source().take(2).toArray(), + every: await source().every(n => n < 4), + forEach, + reduce: await source().reduce((total, n) => total + n, 0), + some: await source().some(n => n === 4), + find: await source().find(n => n > 2), + borrowed: await Readable.prototype.toArray.call((async function* () { yield "a"; })()), + }), + ); + })(); + `); + expect({ ...result, stdout: JSON.parse(result.stdout || "null") }).toEqual({ + stdout: { + drop: [4], + filter: [2, 4], + flatMap: [1, -1, 2, -2, 3, -3, 4, -4], + map: [10, 20, 30, 40], + take: [1, 2], + every: false, + forEach: 10, + reduce: 10, + some: true, + find: 3, + borrowed: ["a"], + }, + stderr: "", + exitCode: 0, + }); + }); + + it.concurrent("bad arguments throw or reject as in Node", async () => { + const result = await run(` + const Readable = require("node:_stream_readable"); + const stream = new Readable({ read() {} }); + const thrown = fn => { + try { + fn(); + return "nothing"; + } catch (error) { + return error.code; + } + }; + let rejected; + const found = { + "map(123) throws": thrown(() => stream.map(123)), + "drop(-1) throws": thrown(() => stream.drop(-1)), + "new map() throws": thrown(() => new stream.map(x => x)), + "new toArray() throws": thrown(() => new stream.toArray()), + "forEach(123) throws": thrown(() => (rejected = stream.forEach(123))), + }; + rejected.catch(error => console.log(JSON.stringify({ ...found, "forEach(123) rejects with": error.code }))); + `); + expect({ ...result, stdout: JSON.parse(result.stdout || "null") }).toEqual({ + stdout: { + "map(123) throws": "ERR_INVALID_ARG_TYPE", + "drop(-1) throws": "ERR_OUT_OF_RANGE", + "new map() throws": "ERR_ILLEGAL_CONSTRUCTOR", + "new toArray() throws": "ERR_ILLEGAL_CONSTRUCTOR", + "forEach(123) throws": "nothing", + "forEach(123) rejects with": "ERR_INVALID_ARG_TYPE", + }, + stderr: "", + exitCode: 0, + }); + }); + + it.concurrent("a global replaced during the first call is not used after it is restored", async () => { + const result = await run(` + const Readable = require("node:_stream_readable"); + const replaced = [ + [Math, "floor"], + [Number, "isNaN"], + [Array.prototype, "push"], + ]; + const callsAfterRestore = {}; + let restored = false; + const originals = replaced.map(([object, key]) => { + const original = object[key]; + callsAfterRestore[key] = 0; + object[key] = function (...args) { + if (restored) callsAfterRestore[key]++; + return Reflect.apply(original, this, args); + }; + return original; + }); + const collect = () => + Readable.from([1.5, 2.5, 3.5]) + .map(n => n * 2, { concurrency: 2 }) + .drop(1) + .take(1) + .toArray(); + collect().then(async first => { + replaced.forEach(([object, key], i) => { + object[key] = originals[i]; + }); + restored = true; + const second = await collect(); + console.log(JSON.stringify({ first, second, callsAfterRestore })); + }); + `); + expect({ ...result, stdout: JSON.parse(result.stdout || "null") }).toEqual({ + stdout: { + first: [5], + second: [5], + callsAfterRestore: { floor: 0, isNaN: 0, push: 0 }, + }, + stderr: "", + exitCode: 0, + }); + }); + + it.concurrent("nothing is defined on Readable.prototype after the class loads", async () => { + const result = await run(` + const Readable = require("node:_stream_readable"); + const { prototype } = Readable; + const mine = () => {}; + prototype.map = mine; + delete prototype.filter; + Object.preventExtensions(prototype); + Readable.from([1]).toArray().then(toArray => { + require("node:stream"); + console.log(JSON.stringify({ toArray, mapIsMine: prototype.map === mine, hasFilter: "filter" in prototype })); + }); + `); + expect(result).toEqual({ + stdout: JSON.stringify({ toArray: [1], mapIsMine: true, hasFilter: false }) + "\n", + stderr: "", + exitCode: 0, + }); + }); + + it.concurrent("map() calls from() of the current stream.Readable, as in Node", async () => { + const result = await run(` + const Readable = require("node:_stream_readable"); + const source = Readable.from([1]); + const calls = []; + const from = Readable.from; + Readable.from = function (iterable) { + calls.push("Readable.from"); + return from.call(this, iterable); + }; + source.map(x => x).destroy(); + Readable.from = from; + + const stream = require("node:stream"); + stream.Readable = { + from(iterable) { + calls.push("a replaced stream.Readable"); + return from.call(Readable, iterable); + }, + }; + source.map(x => x).destroy(); + console.log(JSON.stringify(calls)); + `); + expect(result).toEqual({ + stdout: JSON.stringify(["Readable.from", "a replaced stream.Readable"]) + "\n", + stderr: "", + exitCode: 0, + }); + }); + + it.concurrent("toArray() collects a child's stdout", async () => { + const result = await run(` + const child = require("node:child_process").spawn(process.execPath, ["-e", "console.log('hi')"], { + stdio: ["ignore", "pipe", "ignore"], + }); + child.stdout.toArray().then(chunks => console.log(JSON.stringify(Buffer.concat(chunks).toString()))); + `); + expect(result).toEqual({ stdout: JSON.stringify("hi\n") + "\n", stderr: "", exitCode: 0 }); + }); + + it.concurrent("map() and toArray() collect a net.Socket", async () => { + const result = await run(` + const net = require("node:net"); + const server = net.createServer(socket => socket.end("hello")); + server.listen(0, "127.0.0.1", async () => { + const socket = net.connect(server.address().port, "127.0.0.1"); + const chunks = await socket.map(chunk => chunk.toString().toUpperCase()).toArray(); + server.close(); + console.log(chunks.join("")); + }); + `); + expect(result).toEqual({ stdout: "HELLO\n", stderr: "", exitCode: 0 }); + }); +}); + describe("duplexPair teardown (test-duplex-error.js)", () => { const once = (emitter, event) => new Promise(resolve => emitter.once(event, resolve)); @@ -2041,3 +2364,324 @@ it("Readable.fromWeb's native source accepts pull(view) without the flags array" delete Object.prototype.setFlowing; } }); + +// A built-in module that needs one stream class takes it from the module of that class. node:stream as a whole (the +// operators, pipeline, compose, the other classes) loads when something asks for node:stream. oven-sh/bun#48005. +// node:http is the exception: its classes extend what a program has laid on node:stream when it loads. +describe.concurrent("a built-in module that needs one stream class does not load all of node:stream", () => { + const loadsTheRest = (module, expression) => ` + const m = process.getBuiltinModule(${JSON.stringify(module)}); + let Stream = ${expression}; + while (Stream.name !== "Stream") Stream = Object.getPrototypeOf(Stream); + console.log(typeof Stream.pipeline, typeof Stream.compose, "Readable" in Stream); + `; + it.each([ + ["node:zlib", "m.Gzip"], + ["node:http2", "m.Http2ServerResponse"], + ])("%s", async (module, expression) => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", loadsTheRest(module, expression)], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "undefined undefined false\n", stderr: "" }); + expect(exitCode).toBe(0); + }); +}); + +// A program that has loaded node:stream may store a class or a function of its own under a name there. A built-in module +// takes what lies under the name when that module loads, and never asks again. It does so without loading node:stream. +describe.concurrent("a built-in module uses what a program stored on node:stream before the module loaded", () => { + // around(load) runs `load` with the program's classes stored before it or behind it. + const script = (when, text) => `{ + const stream = require("node:stream"); + const mine = {}; + let finishedCalls = 0; + function store() { + for (const name of ["Stream", "Readable", "Writable", "Duplex", "Transform"]) { + const Was = stream[name]; + function Mine(...given) { + return Was.apply(this, given); + } + Object.setPrototypeOf(Mine.prototype, Was.prototype); + Object.setPrototypeOf(Mine, Was); + stream[name] = mine[name] = Mine; + } + const finished = stream.finished; + stream.finished = function (...given) { + finishedCalls++; + return finished.apply(this, given); + }; + } + function around(load) { + if (${JSON.stringify(when)} === "before") store(); + const loaded = load(); + if (${JSON.stringify(when)} === "after") store(); + return loaded; + } + const print = uses => console.log(JSON.stringify({ uses, finishedCalls })); + ${text} + }`; + // [what loads, what the program asks of it, how often that calls the program's finished(), the program] + const modules = [ + [ + "node:fs and the standard streams", + ["fs.createWriteStream()", "fs.createReadStream()", "process.stdout", "process.stdin"], + 1, + `const nodeFs = around(() => { + const nodeFs = require("node:fs"); + void nodeFs.ReadStream; + void nodeFs.WriteStream; + void process.stdout; + void process.stdin; + return nodeFs; + }); + const readStream = nodeFs.createReadStream(process.execPath); + const writeStream = nodeFs.createWriteStream(require("node:os").devNull); + const uses = { + "fs.createWriteStream()": writeStream instanceof mine.Writable, + "fs.createReadStream()": readStream instanceof mine.Readable, + "process.stdout": process.stdout instanceof mine.Writable, + "process.stdin": process.stdin instanceof mine.Readable, + }; + readStream.close(() => { + writeStream.destroy(); + print(uses); + });`, + ], + [ + "node:zlib", + ["zlib.createGzip()"], + 1, + `const gzip = around(() => require("node:zlib")).createGzip(); + const uses = { "zlib.createGzip()": gzip instanceof mine.Transform }; + gzip.close(() => print(uses));`, + ], + [ + "node:http", + ["http.OutgoingMessage", "the socket of a ServerResponse"], + 0, + `const nodeHttp = around(() => require("node:http")); + const response = new nodeHttp.ServerResponse({ method: "GET", httpVersionMajor: 1, httpVersionMinor: 1 }); + print({ + "http.OutgoingMessage": Object.getPrototypeOf(nodeHttp.OutgoingMessage.prototype) === mine.Stream.prototype, + "the socket of a ServerResponse": response.socket instanceof mine.Duplex, + });`, + ], + [ + "node:http2", + ["http2.Http2ServerRequest"], + 0, + `const nodeHttp2 = around(() => require("node:http2")); + print({ "http2.Http2ServerRequest": Object.getPrototypeOf(nodeHttp2.Http2ServerRequest) === mine.Readable });`, + ], + ]; + it.each( + modules.flatMap(([module, names, finishedCalls, text]) => [ + [module, "before", names, true, finishedCalls, text], + [module, "after", names, false, 0, text], + ]), + )("%s, stored %s", async (module, when, names, used, finishedCalls, text) => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script(when, text)], + env: bunEnv, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ + stdout: JSON.stringify({ uses: Object.fromEntries(names.map(name => [name, used])), finishedCalls }) + "\n", + stderr: "", + }); + expect(exitCode).toBe(0); + }); + + // A test may lay a stub under a name for a while. The fs streams cannot stand below it: their load fails while the + // stub lies there, and runs again once the class is back. + it.each([ + ["an arrow function", "() => {}"], + ["an object", "{}"], + ])("%s under stream.Readable fails the load of the fs streams, which load again behind it", async (_, stub) => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `{ + const stream = require("node:stream"); + const nodeFs = require("node:fs"); + const Readable = stream.Readable; + stream.Readable = ${stub}; + let whileStubbed; + try { + nodeFs.createWriteStream(require("node:os").devNull).destroy(); + whileStubbed = "a stream"; + } catch (error) { + whileStubbed = error.name + ": " + error.message; + } + stream.Readable = Readable; + const readStream = nodeFs.createReadStream(process.execPath); + console.log(JSON.stringify({ whileStubbed, afterwards: readStream instanceof Readable })); + readStream.destroy(); + }`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ + stdout: + JSON.stringify({ whileStubbed: "TypeError: Base class must have a prototype property", afterwards: true }) + + "\n", + stderr: "", + }); + expect(exitCode).toBe(0); + }); + + // A read of process.stdout loads the fs streams. From then on fs.ReadStream stands below the library's own Readable, + // whatever a program stores under stream.Readable later, as when that read loaded all of node:stream. + it.concurrent( + "behind a read of process.stdout, fs.ReadStream does not take what stream.Readable holds at its first use", + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `{ + void process.stdout; + const nodeStream = require("node:stream"); + const nodeFs = require("node:fs"); + const Was = nodeStream.Readable; + nodeStream.Readable = {}; + let whileStubbed; + try { + nodeFs.createReadStream(process.execPath).destroy(); + whileStubbed = "a stream"; + } catch (error) { + whileStubbed = error.name; + } + function Mine(...given) { + return Was.apply(this, given); + } + Object.setPrototypeOf(Mine.prototype, Was.prototype); + Object.setPrototypeOf(Mine, Was); + nodeStream.Readable = Mine; + const readStream = nodeFs.createReadStream(process.execPath); + nodeStream.Readable = Was; + const later = nodeFs.createReadStream(process.execPath); + console.log( + JSON.stringify({ + whileStubbed, + used: readStream instanceof Mine, + above: Object.getPrototypeOf(nodeFs.ReadStream) === Mine, + later: later instanceof Mine, + }), + ); + readStream.destroy(); + later.destroy(); + }`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ + stdout: JSON.stringify({ whileStubbed: "a stream", used: false, above: false, later: false }) + "\n", + stderr: "", + }); + expect(exitCode).toBe(0); + }, + ); + + // The first read of process.stdin is a first use of fs.ReadStream. A stub under stream.Readable then does not cost the program its stdin. + it.concurrent( + "behind a read of process.stdout, a stub under stream.Readable leaves process.stdin a stream", + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `{ + void process.stdout; + const nodeStream = require("node:stream"); + const Was = nodeStream.Readable; + nodeStream.Readable = {}; + const whileStubbed = typeof process.stdin?.on; + nodeStream.Readable = Was; + console.log( + JSON.stringify({ whileStubbed, afterwards: typeof process.stdin?.on, readable: process.stdin instanceof Was }), + ); + }`, + ], + env: bunEnv, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ + stdout: JSON.stringify({ whileStubbed: "function", afterwards: "function", readable: true }) + "\n", + stderr: "", + }); + expect(exitCode).toBe(0); + }, + ); + + // Where no class lies under the name, the fs streams stand below the library's own Readable. + it.each([ + ["null", "stream.Readable = null"], + ["undefined", "stream.Readable = undefined"], + ["deleted", "delete stream.Readable"], + ])("stream.Readable %s: the fs streams stand below the library's own class", async (_, line) => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `{ + const stream = require("node:stream"); + const nodeFs = require("node:fs"); + const Readable = stream.Readable; + ${line}; + const first = nodeFs.createReadStream(process.execPath); + stream.Readable = Readable; + const second = nodeFs.createReadStream(process.execPath); + console.log(JSON.stringify([first instanceof Readable, typeof first.on, second instanceof Readable])); + first.destroy(); + second.destroy(); + }`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: JSON.stringify([true, "function", true]) + "\n", stderr: "" }); + expect(exitCode).toBe(0); + }); + + // node:stream sets its members on `Stream`, the function that every stream class stands on. node:http has loaded all + // of node:stream, so a program that freezes the chain of its classes can still load node:stream. + it("node:stream loads behind a frozen class chain of node:http", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `{ + let above = require("node:http").ServerResponse; + for (; above !== Function.prototype; above = Object.getPrototypeOf(above)) Object.freeze(above); + console.log(typeof require("node:stream").Readable); + }`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "function\n", stderr: "" }); + expect(exitCode).toBe(0); + }); +}); diff --git a/test/js/node/stream/stream-operators-arrive-with-first-pipe.expected.txt b/test/js/node/stream/stream-operators-arrive-with-first-pipe.expected.txt deleted file mode 100644 index 806ef5eaf1a4..000000000000 --- a/test/js/node/stream/stream-operators-arrive-with-first-pipe.expected.txt +++ /dev/null @@ -1,12 +0,0 @@ -nothing: undefined undefined undefined 35 readableAborted,readableHighWaterMark,readableBuffer,readableFlowing,readableLength,readableObjectMode,readableEncoding,errored,closed,destroyed,readableEnded 0 -console.log: undefined undefined undefined 35 readableAborted,readableHighWaterMark,readableBuffer,readableFlowing,readableLength,readableObjectMode,readableEncoding,errored,closed,destroyed,readableEnded 0 -process.stdout: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 -process.stderr: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 -node:fs: undefined undefined undefined 35 readableAborted,readableHighWaterMark,readableBuffer,readableFlowing,readableLength,readableObjectMode,readableEncoding,errored,closed,destroyed,readableEnded 0 -node:http: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 -pipe: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 -pipe, end: false: undefined undefined undefined 35 readableAborted,readableHighWaterMark,readableBuffer,readableFlowing,readableLength,readableObjectMode,readableEncoding,errored,closed,destroyed,readableEnded 0 -pipe, end: true: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 -pipe twice: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 -pipe, both replaced: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 -pipe to a replaced stdout: undefined function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 diff --git a/test/js/node/stream/stream-operators-come-with-the-readable-class.expected.txt b/test/js/node/stream/stream-operators-come-with-the-readable-class.expected.txt new file mode 100644 index 000000000000..272358c37cf0 --- /dev/null +++ b/test/js/node/stream/stream-operators-come-with-the-readable-class.expected.txt @@ -0,0 +1,12 @@ +nothing: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +console.log: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +process.stdout: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +process.stderr: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +node:fs: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +node:http: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +pipe: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +pipe, end: false: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +pipe, end: true: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +pipe twice: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +pipe, both replaced: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0 +pipe to a replaced stdout: function function function 46 drop,filter,flatMap,map,take,every,forEach,reduce,toArray,some,find 0