import {finished} from 'node:stream/promises'; // Wraps `finished(stream)` to handle the following case: // - When the subprocess exits, Node.js automatically calls `subprocess.stdin.destroy()`, which we need to ignore. // - However, we still need to throw if `subprocess.stdin.destroy()` is called before subprocess exit. export const waitForStream = async (stream, fdNumber, streamInfo, {isSameDirection, stopOnExit = false} = {}) => { const state = handleStdinDestroy(stream, streamInfo); const abortController = new AbortController(); try { await Promise.race([ ...(stopOnExit ? [streamInfo.exitPromise] : []), finished(stream, {cleanup: true, signal: abortController.signal}), ]); } catch (error) { if (!state.stdinCleanedUp) { handleStreamError(error, fdNumber, streamInfo, isSameDirection); } } finally { abortController.abort(); } }; // If `subprocess.stdin` is destroyed before being fully written to, it is considered aborted and should throw an error. // This can happen for example when user called `subprocess.stdin.destroy()` before `subprocess.stdin.end()`. // However, Node.js calls `subprocess.stdin.destroy()` on exit for cleanup purposes. // https://github.com/nodejs/node/blob/0b4cdb4b42956cbd7019058e409e06700a199e11/lib/internal/child_process.js#L278 // This is normal and should not throw an error. // Therefore, we need to differentiate between both situations to know whether to throw an error. // Unfortunately, events (`close`, `error`, `end`, `exit`) cannot be used because `.destroy()` can take an arbitrary amount of time. // For example, `stdin: 'pipe'` is implemented as a TCP socket, and its `.destroy()` method waits for TCP disconnection. // Therefore `.destroy()` might end before or after subprocess exit, based on OS speed and load. // The only way to detect this is to spy on `subprocess.stdin._destroy()` by wrapping it. // If `subprocess.exitCode` or `subprocess.signalCode` is set, it means `.destroy()` is being called by Node.js itself. const handleStdinDestroy = (stream, {originalStreams: [originalStdin], subprocess}) => { const state = {stdinCleanedUp: false}; if (stream === originalStdin) { spyOnStdinDestroy(stream, subprocess, state); } return state; }; const spyOnStdinDestroy = (subprocessStdin, subprocess, state) => { const {_destroy} = subprocessStdin; subprocessStdin._destroy = (...destroyArguments) => { setStdinCleanedUp(subprocess, state); _destroy.call(subprocessStdin, ...destroyArguments); }; }; const setStdinCleanedUp = ({exitCode, signalCode}, state) => { if (exitCode !== null || signalCode !== null) { state.stdinCleanedUp = true; } }; // We ignore EPIPEs on writable streams and aborts on readable streams since those can happen normally. // When one stream errors, the error is propagated to the other streams on the same file descriptor. // Those other streams might have a different direction due to the above. // When this happens, the direction of both the initial stream and the others should then be taken into account. // Therefore, we keep track of whether a stream error is currently propagating. const handleStreamError = (error, fdNumber, streamInfo, isSameDirection) => { if (!shouldIgnoreStreamError(error, fdNumber, streamInfo, isSameDirection)) { throw error; } }; const shouldIgnoreStreamError = (error, fdNumber, streamInfo, isSameDirection = true) => { if (streamInfo.propagating) { return isStreamEpipe(error) || isStreamAbort(error); } streamInfo.propagating = true; return isInputFileDescriptor(streamInfo, fdNumber) === isSameDirection ? isStreamEpipe(error) : isStreamAbort(error); }; // Unfortunately, we cannot use the stream's class or properties to know whether it is readable or writable. // For example, `subprocess.stdin` is technically a Duplex, but can only be used as a writable. // Therefore, we need to use the file descriptor's direction (`stdin` is input, `stdout` is output, etc.). // However, while `subprocess.std*` and transforms follow that direction, any stream passed the `std*` option has the opposite direction. // For example, `subprocess.stdin` is a writable, but the `stdin` option is a readable. export const isInputFileDescriptor = ({fileDescriptors}, fdNumber) => fdNumber !== 'all' && fileDescriptors[fdNumber].direction === 'input'; // When `stream.destroy()` is called without an `error` argument, stream is aborted. // This is the only way to abort a readable stream, which can be useful in some instances. // Therefore, we ignore this error on readable streams. export const isStreamAbort = error => error?.code === 'ERR_STREAM_PREMATURE_CLOSE'; // When `stream.write()` is called but the underlying source has been closed, `EPIPE` is emitted. // When piping subprocesses, the source subprocess usually decides when to stop piping. // However, there are some instances when the destination does instead, such as `... | head -n1`. // It notifies the source by using `EPIPE`. // Therefore, we ignore this error on writable streams. const isStreamEpipe = error => error?.code === 'EPIPE';