Release the parallelism slot when a dispatched message rejects

[Ngo Quoc Viet (NgoQuocViet2001)]

Problem

With maxParallelism configured, the message queue's slot accounting lives only in the fulfilment callback:

// jsonrpc/src/common/connection.ts
inFlight++;
...
if (result instanceof Promise) {
    result.then(() => {
        inFlight--;
        triggerMessageQueue();
    }).catch((error) => {
        logger.error(`Processing message queue failed: ${error.toString()}`);
    });
}

Written as .then(onFulfilled).catch(onRejected), a rejection skips onFulfilled entirely. The .catch logs, but never runs inFlight-- and never re-pumps. inFlight is the gate triggerMessageQueue checks:

if (maxParallelism !== -1 && inFlight >= maxParallelism) {
    return;
}

so one rejected dispatch leaves the counter permanently at the limit and the queue stops draining. The connection stays open and Listening; there is no error event, no close, nothing but a single log line, while every later request and notification sits in messageQueue forever.

Trigger

The promise the queue awaits is handleRequest's, which ends in messageWriter.write(...). A response that cannot be serialized rejects it. A handler throwing a ResponseError whose data is not JSON-serializable is enough — toJson() copies data onto the wire message:

server.onRequest(first, () => {
    const circular = {}; circular.self = circular;
    throw new ResponseError(ErrorCodes.InternalError, 'boom', circular);
});
server.onRequest(second, () => 'handled');

With { maxParallelism: 1 }, the request after it is never answered.

A handler that merely throws is not affected — that is caught inside handleRequest and turned into a well-formed replyError. It is the failure to write the response that escapes.

Fix

Use the two-argument form of then, so both settlements do the same bookkeeping. The log line is unchanged.

result.then(() => {
    inFlight--;
    triggerMessageQueue();
}, (error) => {
    inFlight--;
    logger.error(`Processing message queue failed: ${error.toString()}`);
    triggerMessageQueue();
});

The maxParallelism === -1 default never reaches the gate, so unlimited-parallelism behaviour is untouched.

Test plan

• Added Parallelism - a rejected dispatch releases its slot to jsonrpc/src/node/test/connection.test.ts, beside the existing Parallelism - limited case and reusing its TestDuplex pair.
• Ran: npm run compile:jsonrpc then the node suite from jsonrpc/ → 68 passing.
• Checked: reverting only connection.ts fails the new test with 'timed out' — the second request is never answered.
• Ran: npx eslint src in jsonrpc/ → clean.

[azure-pipelines]

Azure Pipelines:
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

🟡 Changes recommended

Error logging can still throw before the queue is restarted, leaving queued messages stalled.

Get a fresh assessment by requesting another Copilot review.

Pull request overview

Fixes queue stalling when a dispatched message rejects under limited parallelism.

Changes:

• Releases the parallelism slot on promise rejection.
• Adds regression coverage for serialization failures.

File summaries

File	Description
jsonrpc/src/common/connection.ts	Handles rejected dispatch bookkeeping.
jsonrpc/src/node/test/connection.test.ts	Tests queue recovery after rejection.

Review details

• Files reviewed: 2/2 changed files
• Comments generated: 1
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Dirk Bäumer (dbaeumer)]

Copilot why not simply

					result.then(() => {
					}).catch((error) => {
						logger.error(`Processing message queue failed: ${error.toString()}`);
					}).finally(() => {
						inFlight--;
						triggerMessageQueue();
					});

[Dirk Bäumer (dbaeumer)]

Agreed, this is simpler and behaves the same: finally always runs on either settlement, so it's the single place to decrement inFlight and call triggerMessageQueue(), while catch only handles the logging.

if (result instanceof Promise) {
	result.then(() => {
	}).catch((error) => {
		logger.error(`Processing message queue failed: ${error.toString()}`);
	}).finally(() => {
		inFlight--;
		triggerMessageQueue();
	});
} else {
	inFlight--;
}

I verified this locally (all existing + new jsonrpc tests pass) and added a second test that forces the dispatched promise to reject via a custom messageStrategy, asserting both that the error is logged and that the parallelism slot is released for the next request:

test('Parallelism - a rejected dispatch is logged and releases its slot', async () => {
	const requestOne = new hostConnection.RequestType0<void, void>('test/parallelism_reject_logged');
	const requestTwo = new hostConnection.RequestType0<string, void>('test/parallelism_after_reject_logged');
	const duplexStream1 = new TestDuplex('ds1');
	const duplexStream2 = new TestDuplex('ds2');

	const errors: string[] = [];
	const logger: hostConnection.Logger = {
		...hostConnection.NullLogger,
		error: (message: string) => errors.push(message)
	};

	// Force the promise the message queue awaits to reject directly, without
	// relying on a serialization failure to produce the rejection.
	const messageStrategy: hostConnection.MessageStrategy = {
		handleMessage: (message, next) => {
			const result = next(message);
			if (hostConnection.Message.isRequest(message) && message.method === requestOne.method) {
				return Promise.reject(new Error('forced rejection'));
			}
			return result;
		}
	};

	const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, logger, { maxParallelism: 1, messageStrategy });
	server.onRequest(requestOne, () => { });
	server.onRequest(requestTwo, () => 'handled');
	server.listen();

	const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { maxParallelism: 1 });
	client.listen();

	await client.sendRequest(requestOne);
	await new Promise(resolve => setTimeout(resolve, 100));

	assert.strictEqual(errors.length, 1);
	assert.ok(errors[0].includes('forced rejection'));

	// The queue must still be pumping, so an ordinary request is answered.
	const answered = await Promise.race([
		client.sendRequest(requestTwo),
		new Promise<string>(resolve => setTimeout(() => resolve('timed out'), 1000))
	]);
	assert.strictEqual(answered, 'handled');
});

I don't have push access to this PR's branch (it's on your fork), so feel free to apply this directly. Happy to open a separate PR instead if that's easier.

[Dirk Bäumer (dbaeumer)]

Ngo Quoc Viet (@NgoQuocViet2001) can you see my latest comment. It comes from Copilot via my user id. I think that change makes sense.