Add nodeChildProcess to subprocess promises and make the subprocess a normal promise (#1255)

Author: sindresorhus

docs/api.md

@@ -211,7 +211,7 @@ This can only be called inside a subprocess. This requires the [`gracefulCancel`
 ## Return value
 
 _TypeScript:_ [`ResultPromise`](typescript.md)\
-_Type:_ `Promise<object> | Subprocess`
+_Type:_ `Subprocess`
 
 The return value of all [asynchronous methods](#methods) is both:
 - the [subprocess](#subprocess).
@@ -223,7 +223,9 @@ The return value of all [asynchronous methods](#methods) is both:
 
 _TypeScript:_ [`Subprocess`](typescript.md)
 
-[`child_process` instance](https://nodejs.org/api/child_process.html#child_process_class_childprocess) with the following methods and properties.
+Promise with the following Execa-specific methods and properties.
+
+The underlying Node.js [`ChildProcess`](https://nodejs.org/api/child_process.html#class-childprocess) instance is available as [`subprocess.nodeChildProcess`](#subprocessnodechildprocess). This is an escape hatch for Node.js-specific APIs that Execa does not document, such as `.on()`, `.send()`, `.disconnect()`, `.ref()` and `.unref()`.
 
 ### subprocess\[Symbol.asyncIterator\]()
 
@@ -335,6 +337,14 @@ This is `undefined` if the subprocess failed to spawn.
 
 [More info.](termination.md#inter-process-termination)
 
+### subprocess.nodeChildProcess
+
+_Type:_ [`ChildProcess`](https://nodejs.org/api/child_process.html#class-childprocess)
+
+Underlying Node.js child process instance.
+
+This is intended as an escape hatch for Node.js-specific APIs that Execa does not document.
+
 ### subprocess.sendMessage(message, sendMessageOptions)
 
 `message`: [`Message`](ipc.md#message-type)\

docs/execution.md

@@ -127,7 +127,7 @@ await timedExeca`npm run test`;
 
 ### Subprocess
 
-The subprocess is returned as soon as it is spawned. It is a [`child_process` instance](https://nodejs.org/api/child_process.html#child_process_class_childprocess) with [additional methods and properties](api.md#subprocess).
+The subprocess is returned as soon as it is spawned. It is a `Promise` with [Execa-specific methods and properties](api.md#subprocess). The underlying Node.js [`ChildProcess`](https://nodejs.org/api/child_process.html#class-childprocess) is available as [`subprocess.nodeChildProcess`](api.md#subprocessnodechildprocess).
 
 ```js
 const subprocess = execa`npm run build`;
@@ -136,7 +136,7 @@ console.log(subprocess.pid);
 
 ### Result
 
-The subprocess is also a `Promise` that resolves with the [`result`](api.md#result).
+The subprocess resolves with the [`result`](api.md#result).
 
 ```js
 const {stdout} = await execa`npm run build`;

lib/convert/duplex.js

@@ -52,7 +52,7 @@ export const createDuplex = ({subprocess, concurrentStreams, encoding}, {from, t
 		subprocess,
 		subprocessStdin,
 	});
-	onStdinFinished(subprocessStdin, duplex, subprocessStdout);
+	onStdinFinished(subprocessStdin, duplex, subprocessStdout, subprocess);
 	return duplex;
 };
 

lib/convert/readable.js

@@ -96,10 +96,24 @@ export const onStdoutFinished = async ({subprocessStdout, onStdoutDataDone, read
 		}
 	} catch (error) {
 		await safeWaitForSubprocessStdin(subprocessStdin);
-		destroyOtherReadable(readable, error);
+		destroyOtherReadable(readable, await getPrematureCloseError(subprocess, error));
 	}
 };
 
+const getPrematureCloseError = async (subprocess, error) => {
+	if (error.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
+		return error;
+	}
+
+	try {
+		await subprocess;
+	} catch (subprocessError) {
+		return subprocessError;
+	}
+
+	return error;
+};
+
 // When `readable` aborts/errors, do the same on `subprocess.stdout`
 export const onReadableDestroy = async ({subprocessStdout, subprocess, waitReadableDestroy}, error) => {
 	if (!await waitForConcurrentStreams(waitReadableDestroy, subprocess)) {

lib/convert/writable.js

@@ -1,6 +1,7 @@
 import {Writable} from 'node:stream';
 import {callbackify} from 'node:util';
 import {getToStream} from '../arguments/fd-options.js';
+import {isStreamAbort} from '../resolve/wait-stream.js';
 import {addConcurrentStream, waitForConcurrentStreams} from './concurrent.js';
 import {
 	safeWaitForSubprocessStdout,
@@ -23,7 +24,7 @@ export const createWritable = ({subprocess, concurrentStreams}, {to} = {}) => {
 		highWaterMark: subprocessStdin.writableHighWaterMark,
 		objectMode: subprocessStdin.writableObjectMode,
 	});
-	onStdinFinished(subprocessStdin, writable);
+	onStdinFinished(subprocessStdin, writable, undefined, subprocess);
 	return writable;
 };
 
@@ -66,18 +67,36 @@ const onWritableFinal = async (subprocessStdin, subprocess, waitWritableFinal) =
 };
 
 // When `subprocess.stdin` ends/aborts/errors, do the same on `writable`.
-export const onStdinFinished = async (subprocessStdin, writable, subprocessStdout) => {
+export const onStdinFinished = async (subprocessStdin, writable, subprocessStdout, subprocess) => {
 	try {
 		await waitForSubprocessStdin(subprocessStdin);
+		await subprocess;
 		if (writable.writable) {
 			writable.end();
 		}
 	} catch (error) {
 		await safeWaitForSubprocessStdout(subprocessStdout);
-		destroyOtherWritable(writable, error);
+		destroyOtherWritable(writable, await getSubprocessError(subprocess, error));
 	}
 };
 
+const getSubprocessError = async (subprocess, error) => {
+	if (!shouldUseSubprocessError(error)) {
+		return error;
+	}
+
+	try {
+		await subprocess;
+	} catch (subprocessError) {
+		return subprocessError;
+	}
+
+	return error;
+};
+
+const shouldUseSubprocessError = error => error === undefined
+	|| isStreamAbort(error);
+
 // When `writable` aborts/errors, do the same on `subprocess.stdin`
 export const onWritableDestroy = async ({subprocessStdin, subprocess, waitWritableFinal, waitWritableDestroy}, error) => {
 	await waitForConcurrentStreams(waitWritableFinal, subprocess);

lib/ipc/forward.js

@@ -51,6 +51,6 @@ const forwardEvents = ({ipcEmitter, anyProcess, channel, isSubprocess}) => {
 export const isConnected = anyProcess => {
 	const ipcEmitter = IPC_EMITTERS.get(anyProcess);
 	return ipcEmitter === undefined
-		? anyProcess.channel !== null
+		? anyProcess.channel !== undefined && anyProcess.channel !== null
 		: ipcEmitter.connected;
 };

lib/ipc/get-each.js

@@ -20,7 +20,7 @@ export const getEachMessage = (subprocessInfo, options = {}) => {
 };
 
 // Same but used internally
-export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAwait, reference, signal}) => {
+export const loopOnMessages = ({anyProcess, waitProcess = anyProcess, channel, isSubprocess, ipc, shouldAwait, reference, signal}) => {
 	validateIpcMethod({
 		methodName: 'getEachMessage',
 		isSubprocess,
@@ -42,6 +42,7 @@ export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAw
 	});
 	return iterateOnMessages({
 		anyProcess,
+		waitProcess,
 		channel,
 		ipcEmitter,
 		isSubprocess,
@@ -83,7 +84,7 @@ const abortOnStrictError = async ({ipcEmitter, isSubprocess, controller, state})
 	} catch {}
 };
 
-const iterateOnMessages = async function * ({anyProcess, channel, ipcEmitter, isSubprocess, shouldAwait, controller, state, reference}) {
+const iterateOnMessages = async function * ({anyProcess, waitProcess, channel, ipcEmitter, isSubprocess, shouldAwait, controller, state, reference}) {
 	try {
 		for await (const [message] of on(ipcEmitter, 'message', {signal: controller.signal})) {
 			throwIfStrictError(state);
@@ -100,7 +101,7 @@ const iterateOnMessages = async function * ({anyProcess, channel, ipcEmitter, is
 		}
 
 		if (shouldAwait) {
-			await anyProcess;
+			await waitProcess;
 		}
 	}
 };

lib/ipc/ipc-input.js

@@ -1,4 +1,5 @@
 import {serialize} from 'node:v8';
+import {sendMessage} from './send.js';
 
 // Validate the `ipcInput` option
 export const validateIpcInputOption = ({ipcInput, ipc, serialization}) => {
@@ -35,10 +36,15 @@ const validateIpcInput = {
 };
 
 // When the `ipcInput` option is set, it is sent as an initial IPC message to the subprocess
-export const sendIpcInput = async (subprocess, ipcInput) => {
+export const sendIpcInput = async (subprocess, ipcInput, ipc) => {
 	if (ipcInput === undefined) {
 		return;
 	}
 
-	await subprocess.sendMessage(ipcInput);
+	await sendMessage({
+		anyProcess: subprocess,
+		channel: subprocess.channel,
+		isSubprocess: false,
+		ipc,
+	}, ipcInput);
 };

lib/ipc/methods.js

@@ -5,8 +5,8 @@ import {getEachMessage} from './get-each.js';
 import {getCancelSignal} from './graceful.js';
 
 // Add promise-based IPC methods in current process
-export const addIpcMethods = (subprocess, {ipc}) => {
-	Object.assign(subprocess, getIpcMethods(subprocess, false, ipc));
+export const addIpcMethods = (target, subprocess, {ipc}) => {
+	Object.assign(target, getIpcMethods(subprocess, false, ipc, target));
 };
 
 // Get promise-based IPC in the subprocess
@@ -27,7 +27,7 @@ export const getIpcExport = () => {
 };
 
 // Retrieve the `ipc` shared by both the current process and the subprocess
-const getIpcMethods = (anyProcess, isSubprocess, ipc) => ({
+const getIpcMethods = (anyProcess, isSubprocess, ipc, waitProcess = anyProcess) => ({
 	sendMessage: sendMessage.bind(undefined, {
 		anyProcess,
 		channel: anyProcess.channel,
@@ -45,5 +45,6 @@ const getIpcMethods = (anyProcess, isSubprocess, ipc) => ({
 		channel: anyProcess.channel,
 		isSubprocess,
 		ipc,
+		waitProcess,
 	}),
 });

lib/ipc/outgoing.js

@@ -1,6 +1,5 @@
 import {createDeferred} from '../utils/deferred.js';
 import {getFdSpecificValue} from '../arguments/specific.js';
-import {SUBPROCESS_OPTIONS} from '../arguments/fd-options.js';
 import {validateStrictDeadlock} from './strict.js';
 
 // When `sendMessage()` is ongoing, any `message` being received waits before being emitted.
@@ -35,13 +34,20 @@ export const waitForOutgoingMessages = async (anyProcess, ipcEmitter, wrappedMes
 };
 
 const OUTGOING_MESSAGES = new WeakMap();
+const IPC_SUBPROCESS_OPTIONS = new WeakMap();
+
+export const setIpcSubprocessOptions = (subprocess, options) => {
+	IPC_SUBPROCESS_OPTIONS.set(subprocess, options);
+};
 
 // Whether any `message` listener is setup
 export const hasMessageListeners = (anyProcess, ipcEmitter) => ipcEmitter.listenerCount('message') > getMinListenerCount(anyProcess);
 
 // When `buffer` is `false`, we set up a `message` listener that should be ignored.
 // That listener is only meant to intercept `strict` acknowledgement responses.
-const getMinListenerCount = anyProcess => SUBPROCESS_OPTIONS.has(anyProcess)
-	&& !getFdSpecificValue(SUBPROCESS_OPTIONS.get(anyProcess).options.buffer, 'ipc')
+const getMinListenerCount = anyProcess => getOptions(anyProcess) !== undefined
+	&& !getFdSpecificValue(getOptions(anyProcess).buffer, 'ipc')
 	? 1
 	: 0;
+
+const getOptions = anyProcess => IPC_SUBPROCESS_OPTIONS.get(anyProcess);