Fix IPC edge cases

Author: sindresorhus

docs/ipc.md

@@ -184,6 +184,8 @@ import {execaNode} from 'execa';
 await execaNode({serialization: 'json'})`child.js`;
 ```
 
+Objects with a `type` property starting with `execa:ipc:` are reserved for Execa's internal messages, and should not be sent.
+
 ## Messages order
 
 The messages are always received in the same order they were sent. Even when sent all at once.

lib/ipc/forward.js

@@ -13,6 +13,8 @@ export const getIpcEmitter = (anyProcess, channel, isSubprocess) => {
 	// Use an `EventEmitter`, like the `process` that is being proxied
 	// eslint-disable-next-line unicorn/prefer-event-target
 	const ipcEmitter = new EventEmitter();
+	// Each pending `getOneMessage()` or `getEachMessage()` call adds listeners, which are removed once it ends. Users cannot reach this emitter to increase its maximum number of listeners.
+	ipcEmitter.setMaxListeners(0);
 	ipcEmitter.connected = true;
 	IPC_EMITTERS.set(anyProcess, ipcEmitter);
 	forwardEvents({

lib/ipc/get-each.js

@@ -56,6 +56,7 @@ export const loopOnMessages = ({anyProcess, waitProcess = anyProcess, channel, i
 		controller,
 		state,
 		reference,
+		signal,
 	});
 };
 
@@ -90,7 +91,7 @@ const abortOnStrictError = async ({ipcEmitter, isSubprocess, controller, state})
 	} catch {}
 };
 
-const iterateOnMessages = async function * ({anyProcess, waitProcess, channel, ipcEmitter, isSubprocess, shouldAwait, controller, state, reference}) {
+const iterateOnMessages = async function * ({anyProcess, waitProcess, channel, ipcEmitter, isSubprocess, shouldAwait, controller, state, reference, signal}) {
 	try {
 		for await (const [message] of on(ipcEmitter, 'message', {signal: controller.signal})) {
 			throwIfStrictError(state);
@@ -102,7 +103,8 @@ const iterateOnMessages = async function * ({anyProcess, waitProcess, channel, i
 		controller.abort();
 		removeReference(channel, reference);
 
-		if (!isSubprocess) {
+		// When the internal `signal` stops the iteration, e.g. when a `.pipe()` is canceled, the subprocess keeps running and can still use IPC
+		if (!isSubprocess && !signal?.aborted) {
 			disconnect(anyProcess);
 		}
 

lib/ipc/get-one.js

@@ -47,7 +47,11 @@ const getOneMessageAsync = async ({anyProcess, channel, isSubprocess, filter, re
 			throwOnStrictError(ipcEmitter, isSubprocess, controller),
 		]);
 	} catch (error) {
-		disconnect(anyProcess);
+		// When the internal `signal` stops the read, e.g. when a `.pipe()` is canceled, the subprocess keeps running and can still use IPC
+		if (!signal?.aborted) {
+			disconnect(anyProcess);
+		}
+
 		throw error;
 	} finally {
 		controller.abort();

lib/ipc/graceful.js

@@ -41,7 +41,8 @@ const startIpc = async ({anyProcess, channel, isSubprocess, ipc}) => {
 
 	isCancelListening = true;
 
-	if (channel === null) {
+	// `channel` is retrieved when Execa is imported, so the current process might have disconnected since then
+	if (!anyProcess.connected) {
 		abortOnDisconnect();
 		return;
 	}

lib/ipc/send.js

@@ -49,6 +49,7 @@ const sendMessageAsync = async ({anyProcess, channel, methodName, isSubprocess,
 			isSubprocess,
 			wrappedMessage,
 			message,
+			strict,
 		});
 	} catch (error) {
 		disconnect(anyProcess);
@@ -59,12 +60,13 @@ const sendMessageAsync = async ({anyProcess, channel, methodName, isSubprocess,
 };
 
 // Used internally by `cancelSignal`
-export const sendOneMessage = async ({anyProcess, methodName, isSubprocess, wrappedMessage, message}) => {
+export const sendOneMessage = async ({anyProcess, methodName, isSubprocess, wrappedMessage, message, strict = false}) => {
 	const sendMethod = getSendMethod(anyProcess);
 
 	try {
 		await Promise.all([
-			waitForStrictResponse(wrappedMessage, anyProcess, isSubprocess),
+			// Only messages sent with the `strict` option wait for an acknowledgment response
+			strict ? waitForStrictResponse(wrappedMessage, anyProcess, isSubprocess) : undefined,
 			sendMethod(wrappedMessage),
 		]);
 	} catch (error) {

lib/ipc/strict.js

@@ -28,7 +28,7 @@ let count = 0;
 // Handles when both processes are calling `sendMessage()` with `strict` at the same time.
 // If neither process is listening, this would create a deadlock. We detect it and throw.
 export const validateStrictDeadlock = (outgoingMessages, wrappedMessage) => {
-	if (wrappedMessage?.type !== REQUEST_TYPE || wrappedMessage.hasListeners) {
+	if (!isStrictRequest(wrappedMessage) || wrappedMessage.hasListeners) {
 		return;
 	}
 
@@ -41,11 +41,16 @@ export const validateStrictDeadlock = (outgoingMessages, wrappedMessage) => {
 
 // The other process then sends the acknowledgment back as a response
 export const handleStrictRequest = async ({wrappedMessage, anyProcess, channel, isSubprocess, ipcEmitter}) => {
-	if (wrappedMessage?.type !== REQUEST_TYPE || !anyProcess.connected) {
+	if (!isStrictRequest(wrappedMessage)) {
 		return wrappedMessage;
 	}
 
 	const {id, message} = wrappedMessage;
+	// The acknowledgment cannot be sent anymore, but the message is still delivered
+	if (!anyProcess.connected) {
+		return message;
+	}
+
 	const response = {id, type: RESPONSE_TYPE, message: hasMessageListeners(anyProcess, ipcEmitter)};
 
 	try {
@@ -56,7 +61,10 @@ export const handleStrictRequest = async ({wrappedMessage, anyProcess, channel,
 			ipc: true,
 		}, response);
 	} catch (error) {
-		ipcEmitter.emit('strict:error', error);
+		// When the other process disconnects right after sending the message, it cannot receive the acknowledgment anymore. Like above, the message is still delivered.
+		if (error.cause?.code !== 'EPIPE') {
+			ipcEmitter.emit('strict:error', error);
+		}
 	}
 
 	return message;
@@ -80,10 +88,6 @@ export const handleStrictResponse = (wrappedMessage, anyProcess) => {
 
 // Wait for the other process to receive the message from `sendMessage()`
 export const waitForStrictResponse = async (wrappedMessage, anyProcess, isSubprocess) => {
-	if (wrappedMessage?.type !== REQUEST_TYPE) {
-		return;
-	}
-
 	const deferred = createDeferred();
 	STRICT_RESPONSES[wrappedMessage.id] = {deferred, anyProcess};
 	const controller = new AbortController();
@@ -117,3 +121,9 @@ const throwOnDisconnect = async (anyProcess, isSubprocess, {signal}) => {
 
 const REQUEST_TYPE = 'execa:ipc:request';
 const RESPONSE_TYPE = 'execa:ipc:response';
+
+// `id` and `hasListeners` are always defined in messages sent by `sendMessage()` with the `strict` option.
+// Messages with the same `type` but without them are user messages, which must be delivered as is.
+const isStrictRequest = wrappedMessage => wrappedMessage?.type === REQUEST_TYPE
+	&& Number.isInteger(wrappedMessage.id)
+	&& typeof wrappedMessage.hasListeners === 'boolean';

lib/ipc/validation.js

@@ -1,3 +1,5 @@
+import {inspect} from 'node:util';
+
 // Validate the IPC channel is connected before receiving/sending messages
 export const validateIpcMethod = ({methodName, isSubprocess, ipc, isConnected}) => {
 	validateIpcOption(methodName, isSubprocess, ipc);
@@ -68,7 +70,16 @@ export const handleEpipeError = ({error, methodName, isSubprocess}) => {
 // Works with both `serialization: 'advanced'` and `serialization: 'json'`.
 export const handleSerializationError = ({error, methodName, isSubprocess, message}) => {
 	if (isSerializationError(error)) {
-		throw new Error(`${getMethodName(methodName, isSubprocess)}'s argument type is invalid: the message cannot be serialized: ${String(message)}.`, {cause: error});
+		throw new Error(`${getMethodName(methodName, isSubprocess)}'s argument type is invalid: the message cannot be serialized: ${serializeInvalidMessage(message)}.`, {cause: error});
+	}
+};
+
+// `String()` throws with objects that have a `null` prototype or a throwing `toString()` method
+const serializeInvalidMessage = message => {
+	try {
+		return String(message);
+	} catch {
+		return inspect(message);
 	}
 };
 

test/fixtures/graceful-print-disconnected.js

@@ -0,0 +1,12 @@
+#!/usr/bin/env node
+import process from 'node:process';
+import {once} from 'node:events';
+import {onAbortedSignal} from '../helpers/graceful.js';
+import {getCancelSignal} from 'execa';
+
+// Unlike `graceful-print.js`, Execa is imported before the parent process disconnects
+process.send('.');
+await once(process, 'disconnect');
+const cancelSignal = await getCancelSignal();
+await onAbortedSignal(cancelSignal);
+console.log(cancelSignal.reason);

test/fixtures/ipc-get-concurrent.js

@@ -0,0 +1,6 @@
+#!/usr/bin/env node
+import {argv} from 'node:process';
+import {getOneMessage} from '../../index.js';
+
+const messages = await Promise.all(Array.from({length: Number(argv[2])}, () => getOneMessage()));
+console.log(messages.length);