Handle exceptions from the filter option of getOneMessage()

lib/ipc/get-one.js

@@ -1,5 +1,5 @@
 import {once, on} from 'node:events';
-import {validateIpcMethod, throwOnEarlyDisconnect} from './validation.js';
+import {validateIpcMethod, throwOnEarlyDisconnect, disconnect} from './validation.js';
 import {getIpcEmitter, isConnected} from './forward.js';
 import {addReference, removeReference} from './reference.js';
 
@@ -24,6 +24,9 @@ const getOneMessageAsync = async (anyProcess, isSubprocess, filter) => {
 			getMessage(ipcEmitter, filter, controller),
 			throwOnDisconnect(ipcEmitter, isSubprocess, controller),
 		]);
+	} catch (error) {
+		disconnect(anyProcess);
+		throw error;
 	} finally {
 		controller.abort();
 		removeReference(anyProcess);

test/fixtures/ipc-get-filter-throw.js

@@ -0,0 +1,9 @@
+#!/usr/bin/env node
+import {getOneMessage} from '../../index.js';
+import {foobarString} from '../helpers/input.js';
+
+await getOneMessage({
+	filter() {
+		throw new Error(foobarString);
+	},
+});

test/fixtures/ipc-send-get.js

@@ -0,0 +1,8 @@
+#!/usr/bin/env node
+import {sendMessage, getOneMessage} from '../../index.js';
+import {foobarString} from '../helpers/input.js';
+
+await Promise.all([
+	getOneMessage(),
+	sendMessage(foobarString),
+]);

test/ipc/get-one.js

@@ -57,6 +57,35 @@ test('exports.getOneMessage() can filter messages', async t => {
 	t.deepEqual(ipcOutput, [foobarArray[1]]);
 });
 
+test('Throwing from subprocess.getOneMessage() filter disconnects', async t => {
+	const subprocess = execa('ipc-send-get.js', {ipc: true});
+	const error = new Error(foobarString);
+	t.is(await t.throwsAsync(subprocess.getOneMessage({
+		filter() {
+			throw error;
+		},
+	})), error);
+
+	const {exitCode, isTerminated, message, ipcOutput} = await t.throwsAsync(subprocess);
+	t.is(exitCode, 1);
+	t.false(isTerminated);
+	t.true(message.includes('Error: getOneMessage() could not complete'));
+	t.deepEqual(ipcOutput, [foobarString]);
+});
+
+test('Throwing from exports.getOneMessage() filter disconnects', async t => {
+	const subprocess = execa('ipc-get-filter-throw.js', {ipc: true, ipcInput: 0});
+	await t.throwsAsync(subprocess.getOneMessage(), {
+		message: /subprocess.getOneMessage\(\) could not complete/,
+	});
+
+	const {exitCode, isTerminated, message, ipcOutput} = await t.throwsAsync(subprocess);
+	t.is(exitCode, 1);
+	t.false(isTerminated);
+	t.true(message.includes(`Error: ${foobarString}`));
+	t.deepEqual(ipcOutput, []);
+});
+
 test.serial('Can retrieve initial IPC messages under heavy load', async t => {
 	await Promise.all(
 		Array.from({length: PARALLEL_COUNT}, async (_, index) => {