From b11b9cfb9f549d72abe334f148c74f28c7c0c2ab Mon Sep 17 00:00:00 2001 From: kwy404 Date: Tue, 27 Oct 2026 13:09:55 -0300 Subject: [PATCH 1/7] Add outgoing and incoming message hooks to MessageStrategy A message strategy can now implement handleOutgoingMessage to see or change every message before the connection writes it, and handleIncomingMessage to handle received messages instead of handleMessage. This lets callers add properties like a W3C trace context to the messages they send and read them on the other side. handleMessage keeps handling received messages only, so existing strategies behave as before, and is now optional. Closes #1757 --- jsonrpc/README.md | 16 +++++ jsonrpc/src/common/connection.ts | 51 +++++++++++--- jsonrpc/src/node/test/connection.test.ts | 86 ++++++++++++++++++++++++ 3 files changed, 145 insertions(+), 8 deletions(-) diff --git a/jsonrpc/README.md b/jsonrpc/README.md index 4ffbb87d1..2c0293fbf 100644 --- a/jsonrpc/README.md +++ b/jsonrpc/README.md @@ -50,6 +50,22 @@ connection.onNotification(notification, (param: string) => { connection.listen(); ``` +Both sides can pass a `messageStrategy` in the connection options to work with the messages of the connection. `handleOutgoingMessage` gets every message before it is written, and `handleIncomingMessage` gets every received message before it is dispatched. For example, to send a trace context along with each message and restore it while the other side handles the message (`getTraceparent` and `runWithTraceparent` stand for your own tracing code): + +```ts +let connection = rpc.createMessageConnection(reader, writer, undefined, { + messageStrategy: { + handleOutgoingMessage(message, next) { + return next({ ...message, traceparent: getTraceparent() } as rpc.Message); + }, + handleIncomingMessage(message, next) { + const traceparent = (message as { traceparent?: string }).traceparent; + return runWithTraceparent(traceparent, () => next(message)); + } + } +}); +``` + # History For the history please see the [main repository](https://github.com/Microsoft/vscode-languageserver-node/blob/master/README.md) diff --git a/jsonrpc/src/common/connection.ts b/jsonrpc/src/common/connection.ts index 06bdd5d6c..78cabde6c 100644 --- a/jsonrpc/src/common/connection.ts +++ b/jsonrpc/src/common/connection.ts @@ -474,13 +474,38 @@ export namespace CancellationStrategy { } export interface MessageStrategy { - handleMessage(message: Message, next: (message: Message) => NotificationResult): NotificationResult; + /** + * Handles a received message before it is dispatched. Not called if + * `handleIncomingMessage` is provided. + * + * @param message The received message. + * @param next Dispatches the message. + */ + handleMessage?(message: Message, next: (message: Message) => NotificationResult): NotificationResult; + + /** + * Handles a received message before it is dispatched. If provided, it is + * used instead of `handleMessage`. + * + * @param message The received message. + * @param next Dispatches the message. + */ + handleIncomingMessage?(message: Message, next: (message: Message) => NotificationResult): NotificationResult; + + /** + * Handles a message before it is written, for example to add a property + * like a trace context to it. + * + * @param message The message to send. + * @param next Writes the message. Pass it the message to write. + */ + handleOutgoingMessage?(message: Message, next: (message: Message) => Promise): Promise; } export namespace MessageStrategy { export function is(value: any): value is MessageStrategy { const candidate: MessageStrategy = value; - return candidate && Is.func(candidate.handleMessage); + return candidate && (Is.func(candidate.handleMessage) || Is.func(candidate.handleIncomingMessage) || Is.func(candidate.handleOutgoingMessage)); } } @@ -735,7 +760,9 @@ export function createMessageConnection(messageReader: MessageReader, messageWri try { inFlight++; const messageStrategy = options?.messageStrategy; - if (MessageStrategy.is(messageStrategy)) { + if (messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage)) { + result = messageStrategy.handleIncomingMessage(message, handleMessage); + } else if (messageStrategy !== undefined && Is.func(messageStrategy.handleMessage)) { result = messageStrategy.handleMessage(message, handleMessage); } else { result = handleMessage(message); @@ -775,6 +802,14 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } } + function writeMessage(message: Message): Promise { + const messageStrategy = options?.messageStrategy; + if (messageStrategy !== undefined && Is.func(messageStrategy.handleOutgoingMessage)) { + return messageStrategy.handleOutgoingMessage(message, (message) => messageWriter.write(message)); + } + return messageWriter.write(message); + } + const callback: DataCallback = (message) => { try { // We have received a cancellation message. Check if the message is still in the queue @@ -791,7 +826,7 @@ export function createMessageConnection(messageReader: MessageReader, messageWri requestTokens.delete(cancelId); response.id = toCancel.id; traceSendingResponse(response, message.method, Date.now()); - messageWriter.write(response).catch(() => logger.error(`Sending response for canceled message failed.`)); + writeMessage(response).catch(() => logger.error(`Sending response for canceled message failed.`)); return; } } @@ -831,7 +866,7 @@ export function createMessageConnection(messageReader: MessageReader, messageWri message.result = resultOrError === undefined ? null : resultOrError; } traceSendingResponse(message, method, startTime); - return messageWriter.write(message); + return writeMessage(message); } function replyError(error: ResponseError, method: string, startTime: number): Promise { const message: ResponseMessage = { @@ -840,7 +875,7 @@ export function createMessageConnection(messageReader: MessageReader, messageWri error: error.toJson() }; traceSendingResponse(message, method, startTime); - return messageWriter.write(message); + return writeMessage(message); } traceReceivedRequest(requestMessage); @@ -1313,7 +1348,7 @@ export function createMessageConnection(messageReader: MessageReader, messageWri params: messageParams }; traceSendingNotification(notificationMessage); - return messageWriter.write(notificationMessage).catch((error) => { + return writeMessage(notificationMessage).catch((error) => { logger.error(`Sending notification failed.`); throw error; }); @@ -1460,7 +1495,7 @@ export function createMessageConnection(messageReader: MessageReader, messageWri const responsePromise: ResponsePromise | null = { method: method, timerStart: Date.now(), resolve: resolveWithCleanup, reject: rejectWithCleanup }; try { responsePromises.set(id, responsePromise); - await messageWriter.write(requestMessage); + await writeMessage(requestMessage); if (tokenWasCancelled) { sendCancellation(connection, id); } diff --git a/jsonrpc/src/node/test/connection.test.ts b/jsonrpc/src/node/test/connection.test.ts index 3de33d49a..735cacd28 100644 --- a/jsonrpc/src/node/test/connection.test.ts +++ b/jsonrpc/src/node/test/connection.test.ts @@ -885,6 +885,92 @@ suite('Connection', () => { }); }); + test('Message strategy handles outgoing and incoming messages', async () => { + const type = new RequestType('test/handleSingleRequest'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const asyncLocalStorage = new AsyncLocalStorage(); + + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger, { + messageStrategy: { + handleIncomingMessage(message, next) { + const traceparent = (message as { traceparent?: string }).traceparent; + return asyncLocalStorage.run(traceparent, () => next(message)); + } + } + }); + server.onRequest(type, () => asyncLocalStorage.getStore()); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { + messageStrategy: { + handleOutgoingMessage(message, next) { + return next({ ...message, traceparent: '00-trace-span-01' } as hostConnection.Message); + } + } + }); + client.listen(); + + assert.strictEqual(await client.sendRequest(type, 0), '00-trace-span-01'); + }); + + test('Message strategy handles outgoing responses and notifications', async () => { + const type = new RequestType('test/handleSingleRequest'); + const notificationType = new NotificationType('test/notification'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const sent: string[] = []; + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger, { + messageStrategy: { + handleOutgoingMessage(message, next) { + sent.push(hostConnection.Message.isNotification(message) ? message.method : 'response'); + return next(message); + } + } + }); + server.onRequest(type, async (param) => { + await server.sendNotification(notificationType, param); + return param + 1; + }); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger); + client.listen(); + + assert.strictEqual(await client.sendRequest(type, 1), 2); + assert.deepStrictEqual(sent, ['test/notification', 'response']); + }); + + test('Message strategy prefers handleIncomingMessage over handleMessage', async () => { + const type = new RequestType('test/handleSingleRequest'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const calls: string[] = []; + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger, { + messageStrategy: { + handleMessage(message, next) { + calls.push('handleMessage'); + return next(message); + }, + handleIncomingMessage(message, next) { + calls.push('handleIncomingMessage'); + return next(message); + } + } + }); + server.onRequest(type, (param) => param + 1); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger); + client.listen(); + + assert.strictEqual(await client.sendRequest(type, 1), 2); + assert.deepStrictEqual(calls, ['handleIncomingMessage']); + }); + test('Message ports', async () => { const type = new RequestType('test/handleSingleRequest'); const messageChannel = new MessageChannel(); From ce58efa03c483a6a62e2268abf2539b6c425d011 Mon Sep 17 00:00:00 2001 From: kwy404 Date: Tue, 27 Oct 2026 15:32:17 -0300 Subject: [PATCH 2/7] Keep outgoing order and pass unqueued messages to handleIncomingMessage handleOutgoingMessage now gets one message at a time in send order. A hook that awaited before calling next let a later message reach the writer first, so for example didChange could be written before didOpen. Connections without the hook still write directly. handleIncomingMessage now also gets the messages that skip the queue: responses when maxParallelism is set and $/cancelRequest notifications handled as soon as they are read. They still skip the queue, and handleMessage still does not get them. --- jsonrpc/README.md | 2 +- jsonrpc/src/common/connection.ts | 43 +++++++--- jsonrpc/src/node/test/connection.test.ts | 105 +++++++++++++++++++++++ 3 files changed, 139 insertions(+), 11 deletions(-) diff --git a/jsonrpc/README.md b/jsonrpc/README.md index 2c0293fbf..ba6160b0a 100644 --- a/jsonrpc/README.md +++ b/jsonrpc/README.md @@ -50,7 +50,7 @@ connection.onNotification(notification, (param: string) => { connection.listen(); ``` -Both sides can pass a `messageStrategy` in the connection options to work with the messages of the connection. `handleOutgoingMessage` gets every message before it is written, and `handleIncomingMessage` gets every received message before it is dispatched. For example, to send a trace context along with each message and restore it while the other side handles the message (`getTraceparent` and `runWithTraceparent` stand for your own tracing code): +Both sides can pass a `messageStrategy` in the connection options to work with the messages of the connection. `handleOutgoingMessage` gets every message before it is written, and `handleIncomingMessage` gets every received message before it is dispatched. Outgoing messages are passed one at a time in send order, so a `handleOutgoingMessage` that awaits before calling `next` delays the messages after it instead of reordering them. For example, to send a trace context along with each message and restore it while the other side handles the message (`getTraceparent` and `runWithTraceparent` stand for your own tracing code): ```ts let connection = rpc.createMessageConnection(reader, writer, undefined, { diff --git a/jsonrpc/src/common/connection.ts b/jsonrpc/src/common/connection.ts index 78cabde6c..d7fc98d7a 100644 --- a/jsonrpc/src/common/connection.ts +++ b/jsonrpc/src/common/connection.ts @@ -485,7 +485,9 @@ export interface MessageStrategy { /** * Handles a received message before it is dispatched. If provided, it is - * used instead of `handleMessage`. + * used instead of `handleMessage`. Unlike `handleMessage`, it also gets the + * messages that skip the message queue: responses when `maxParallelism` is + * set and `$/cancelRequest` notifications handled as soon as they are read. * * @param message The received message. * @param next Dispatches the message. @@ -494,7 +496,9 @@ export interface MessageStrategy { /** * Handles a message before it is written, for example to add a property - * like a trace context to it. + * like a trace context to it. Messages are passed one at a time in send + * order: the next one is passed when the promise returned for the previous + * one settles. * * @param message The message to send. * @param next Writes the message. Pass it the message to write. @@ -733,7 +737,7 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } else { // If we have limited parallelism we resolve responses to avoid // dead locks. - handleResponse(message); + handleUnqueuedMessage(message, handleMessage); } } else { queue.set(createNotificationQueueKey(), message); @@ -802,12 +806,27 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } } + // Messages that skip the queue go through `handleIncomingMessage`. + // `handleMessage` keeps getting queued messages only. + function handleUnqueuedMessage(message: Message, handle: (message: Message) => NotificationResult): void { + const messageStrategy = options?.messageStrategy; + const result = messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage) ? messageStrategy.handleIncomingMessage(message, handle) : handle(message); + if (result instanceof Promise) { + result.catch((error) => logger.error(`Processing message failed: ${error.toString()}`)); + } + } + + let lastOutgoingMessage: Promise = Promise.resolve(); function writeMessage(message: Message): Promise { const messageStrategy = options?.messageStrategy; - if (messageStrategy !== undefined && Is.func(messageStrategy.handleOutgoingMessage)) { - return messageStrategy.handleOutgoingMessage(message, (message) => messageWriter.write(message)); + if (messageStrategy === undefined || !Is.func(messageStrategy.handleOutgoingMessage)) { + return messageWriter.write(message); } - return messageWriter.write(message); + // Pass one message at a time to the hook. Otherwise a hook that awaits + // before calling next lets a later message reach the writer first. + const result = lastOutgoingMessage.then(() => messageStrategy.handleOutgoingMessage!(message, (message) => messageWriter.write(message))); + lastOutgoingMessage = result.catch(() => undefined); + return result; } const callback: DataCallback = (message) => { @@ -825,16 +844,20 @@ export function createMessageConnection(messageReader: MessageReader, messageWri messageQueue.delete(key); requestTokens.delete(cancelId); response.id = toCancel.id; - traceSendingResponse(response, message.method, Date.now()); - writeMessage(response).catch(() => logger.error(`Sending response for canceled message failed.`)); + handleUnqueuedMessage(message, () => { + traceSendingResponse(response, message.method, Date.now()); + return writeMessage(response).catch(() => logger.error(`Sending response for canceled message failed.`)); + }); return; } } const cancellationToken = requestTokens.get(cancelId); // The request is already running. Cancel the token if (cancellationToken !== undefined) { - cancellationToken.cancel(); - traceReceivedNotification(message); + handleUnqueuedMessage(message, () => { + cancellationToken.cancel(); + traceReceivedNotification(message); + }); return; } else { // Remember the cancel but still queue the message to diff --git a/jsonrpc/src/node/test/connection.test.ts b/jsonrpc/src/node/test/connection.test.ts index 735cacd28..a8c74d94a 100644 --- a/jsonrpc/src/node/test/connection.test.ts +++ b/jsonrpc/src/node/test/connection.test.ts @@ -971,6 +971,111 @@ suite('Connection', () => { assert.deepStrictEqual(calls, ['handleIncomingMessage']); }); + test('Message strategy keeps the send order when handleOutgoingMessage awaits', async () => { + const didOpen = new NotificationType('test/didOpen'); + const didChange = new NotificationType('test/didChange'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const received: string[] = []; + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger); + const bothReceived = new Promise((resolve) => { + server.onNotification((method) => { + received.push(method); + if (received.length === 2) { + resolve(); + } + }); + }); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { + messageStrategy: { + async handleOutgoingMessage(message, next) { + if (hostConnection.Message.isNotification(message) && message.method === didOpen.method) { + await new Promise(resolve => setTimeout(resolve, 10)); + } + return next(message); + } + } + }); + client.listen(); + + void client.sendNotification(didOpen, 1); + void client.sendNotification(didChange, 2); + await bothReceived; + assert.deepStrictEqual(received, [didOpen.method, didChange.method]); + }); + + test('Message strategy handles responses that skip the queue', async () => { + const type = new RequestType('test/handleSingleRequest'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger); + server.onRequest(type, (param) => param + 1); + server.listen(); + + // With maxParallelism set, responses are handled as soon as they are read. + const received: hostConnection.Message[] = []; + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { + maxParallelism: 1, + messageStrategy: { + handleIncomingMessage(message, next) { + received.push(message); + return next(message); + } + } + }); + client.listen(); + + assert.strictEqual(await client.sendRequest(type, 1), 2); + assert.deepStrictEqual(received, [{ jsonrpc: '2.0', id: 0, result: 2 }]); + }); + + test('Message strategy handles cancellations that skip the queue', async () => { + const type = new RequestType('test/handleSingleRequest'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const received: string[] = []; + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger, { + maxParallelism: 1, + connectionStrategy: { + cancelUndispatched: (message) => ({ jsonrpc: '2.0', id: (message as hostConnection.RequestMessage).id, result: 0 }) + }, + messageStrategy: { + handleIncomingMessage(message, next) { + received.push((message as hostConnection.RequestMessage).method); + return next(message); + } + } + }); + let started!: () => void; + const firstStarted = new Promise(resolve => started = resolve); + server.onRequest(type, (param, token) => { + started(); + return new Promise(resolve => token.onCancellationRequested(() => resolve(param))); + }); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger); + client.listen(); + + // The first request runs while the second one waits in the queue. Both + // cancellations are handled as soon as they are read. + const firstSource = new CancellationTokenSource(); + const first = client.sendRequest(type, 1, firstSource.token); + await firstStarted; + const secondSource = new CancellationTokenSource(); + const second = client.sendRequest(type, 2, secondSource.token); + secondSource.cancel(); + assert.strictEqual(await second, 0); + firstSource.cancel(); + assert.strictEqual(await first, 1); + assert.deepStrictEqual(received, [type.method, '$/cancelRequest', '$/cancelRequest']); + }); + test('Message ports', async () => { const type = new RequestType('test/handleSingleRequest'); const messageChannel = new MessageChannel(); From efce8f7925a26572c2f60d3ad29c4d907d0e0c7a Mon Sep 17 00:00:00 2001 From: kwy404 Date: Tue, 27 Oct 2026 15:46:41 -0300 Subject: [PATCH 3/7] Catch hook errors and defer queued cancellations until next is called A synchronous exception from handleIncomingMessage on a message that skips the queue escaped to the reader. The IPC reader calls the connection from the process message event, so this could crash the process. Such errors are now logged like asynchronous rejections. When cancelUndispatched answered a queued request, the request was removed from the queue before the hook ran, so a hook that never called next left it unanswered. The request is now removed when next is called, and only if it is still queued then. If it already started, its token is cancelled instead, so it is still answered once. --- jsonrpc/src/common/connection.ts | 25 ++++++-- jsonrpc/src/node/test/connection.test.ts | 74 ++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 6 deletions(-) diff --git a/jsonrpc/src/common/connection.ts b/jsonrpc/src/common/connection.ts index d7fc98d7a..0cf32fa75 100644 --- a/jsonrpc/src/common/connection.ts +++ b/jsonrpc/src/common/connection.ts @@ -807,12 +807,18 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } // Messages that skip the queue go through `handleIncomingMessage`. - // `handleMessage` keeps getting queued messages only. + // `handleMessage` keeps getting queued messages only. Errors are logged, + // since the reader can call us from an event handler. function handleUnqueuedMessage(message: Message, handle: (message: Message) => NotificationResult): void { const messageStrategy = options?.messageStrategy; - const result = messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage) ? messageStrategy.handleIncomingMessage(message, handle) : handle(message); - if (result instanceof Promise) { - result.catch((error) => logger.error(`Processing message failed: ${error.toString()}`)); + const logError = (error: any) => logger.error(`Processing message failed: ${error}`); + try { + const result = messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage) ? messageStrategy.handleIncomingMessage(message, handle) : handle(message); + if (result instanceof Promise) { + result.catch(logError); + } + } catch (error) { + logError(error); } } @@ -841,10 +847,17 @@ export function createMessageConnection(messageReader: MessageReader, messageWri const strategy = options?.connectionStrategy; const response = (strategy && strategy.cancelUndispatched) ? strategy.cancelUndispatched(toCancel, cancelUndispatched) : cancelUndispatched(toCancel); if (response && (response.error !== undefined || response.result !== undefined)) { - messageQueue.delete(key); - requestTokens.delete(cancelId); response.id = toCancel.id; handleUnqueuedMessage(message, () => { + // The hook can call next later or not at all. Only drop the + // request if it is still queued then, otherwise cancel it. + if (messageQueue.get(key) !== toCancel) { + requestTokens.get(cancelId)?.cancel(); + traceReceivedNotification(message); + return; + } + messageQueue.delete(key); + requestTokens.delete(cancelId); traceSendingResponse(response, message.method, Date.now()); return writeMessage(response).catch(() => logger.error(`Sending response for canceled message failed.`)); }); diff --git a/jsonrpc/src/node/test/connection.test.ts b/jsonrpc/src/node/test/connection.test.ts index a8c74d94a..2b9b27a70 100644 --- a/jsonrpc/src/node/test/connection.test.ts +++ b/jsonrpc/src/node/test/connection.test.ts @@ -1076,6 +1076,80 @@ suite('Connection', () => { assert.deepStrictEqual(received, [type.method, '$/cancelRequest', '$/cancelRequest']); }); + test('Message strategy can ignore the cancellation of a queued request', async () => { + const type = new RequestType('test/handleSingleRequest'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + let dropped!: () => void; + const cancelDropped = new Promise(resolve => dropped = resolve); + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger, { + maxParallelism: 1, + connectionStrategy: { + cancelUndispatched: (message) => ({ jsonrpc: '2.0', id: (message as hostConnection.RequestMessage).id, result: 0 }) + }, + messageStrategy: { + handleIncomingMessage(message, next) { + // Cancellations are dropped without calling next. + if (hostConnection.Message.isNotification(message) && message.method === '$/cancelRequest') { + dropped(); + return; + } + return next(message); + } + } + }); + let started!: () => void; + const firstStarted = new Promise(resolve => started = resolve); + let release!: () => void; + const released = new Promise(resolve => release = resolve); + server.onRequest(type, async (param) => { + if (param === 1) { + started(); + await released; + } + return param; + }); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger); + client.listen(); + + // The second request waits in the queue while the first one runs. + const first = client.sendRequest(type, 1); + await firstStarted; + const source = new CancellationTokenSource(); + const second = client.sendRequest(type, 2, source.token); + source.cancel(); + await cancelDropped; + release(); + assert.strictEqual(await first, 1); + assert.strictEqual(await second, 2); + }); + + test('Message strategy errors on messages that skip the queue are logged', () => { + const proc = new EventEmitter(); + const errors: string[] = []; + const logger: hostConnection.Logger = { + ...hostConnection.NullLogger, + error: (message: string) => errors.push(message) + }; + const connection = hostConnection.createMessageConnection(new hostConnection.IPCMessageReader(proc as NodeJS.Process), new hostConnection.IPCMessageWriter(proc as NodeJS.Process), logger, { + maxParallelism: 1, + messageStrategy: { + handleIncomingMessage() { + throw new Error('hook failed'); + } + } + }); + connection.listen(); + + // The IPC reader calls the connection from the process message event, + // so an error escaping there would be an uncaught exception. + assert.doesNotThrow(() => proc.emit('message', { jsonrpc: '2.0', id: 0, result: 1 })); + assert.deepStrictEqual(errors, ['Processing message failed: Error: hook failed']); + }); + test('Message ports', async () => { const type = new RequestType('test/handleSingleRequest'); const messageChannel = new MessageChannel(); From 4acc14364b66f62a9979b97eedb0c21e3538ae66 Mon Sep 17 00:00:00 2001 From: kwy404 Date: Thu, 29 Oct 2026 13:17:26 -0300 Subject: [PATCH 4/7] Defer all cancellation bookkeeping until handleIncomingMessage calls next With the default connection strategy, a $/cancelRequest for a queued request was added to knownCanceledRequests before the hook saw it, so the request got a canceled token even when the hook dropped the cancellation. The hook now gets every $/cancelRequest as soon as it is read, and the existing cancellation code only runs when it calls next. The copy that stays queued to clean up is not passed to the hook again. --- jsonrpc/src/common/connection.ts | 48 +++++++++++++----------- jsonrpc/src/node/test/connection.test.ts | 45 ++++++++++++++++++++++ 2 files changed, 72 insertions(+), 21 deletions(-) diff --git a/jsonrpc/src/common/connection.ts b/jsonrpc/src/common/connection.ts index 0cf32fa75..14b496560 100644 --- a/jsonrpc/src/common/connection.ts +++ b/jsonrpc/src/common/connection.ts @@ -487,7 +487,8 @@ export interface MessageStrategy { * Handles a received message before it is dispatched. If provided, it is * used instead of `handleMessage`. Unlike `handleMessage`, it also gets the * messages that skip the message queue: responses when `maxParallelism` is - * set and `$/cancelRequest` notifications handled as soon as they are read. + * set, and `$/cancelRequest` notifications as soon as they are read. A + * request is only canceled once `next` is called for its `$/cancelRequest`. * * @param message The received message. * @param next Dispatches the message. @@ -765,7 +766,8 @@ export function createMessageConnection(messageReader: MessageReader, messageWri inFlight++; const messageStrategy = options?.messageStrategy; if (messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage)) { - result = messageStrategy.handleIncomingMessage(message, handleMessage); + // A queued cancellation already went through the hook when it was read. + result = isCancelNotification(message) ? handleMessage(message) : messageStrategy.handleIncomingMessage(message, handleMessage); } else if (messageStrategy !== undefined && Is.func(messageStrategy.handleMessage)) { result = messageStrategy.handleMessage(message, handleMessage); } else { @@ -836,10 +838,25 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } const callback: DataCallback = (message) => { + const messageStrategy = options?.messageStrategy; + // The hook gets cancellations as soon as they are read and can drop them, + // so nothing is canceled or remembered before it calls next. + if (isCancelNotification(message) && messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage)) { + handleUnqueuedMessage(message, receiveMessage); + } else { + receiveMessage(message); + } + }; + + function isCancelNotification(message: Message): message is NotificationMessage { + return Message.isNotification(message) && message.method === CancelNotification.type.method; + } + + function receiveMessage(message: Message): void { try { // We have received a cancellation message. Check if the message is still in the queue // and cancel it if allowed to do so. - if (Message.isNotification(message) && message.method === CancelNotification.type.method) { + if (isCancelNotification(message)) { const cancelId = (message.params as CancelParams).id; const key = createRequestQueueKey(cancelId); const toCancel = messageQueue.get(key); @@ -847,30 +864,19 @@ export function createMessageConnection(messageReader: MessageReader, messageWri const strategy = options?.connectionStrategy; const response = (strategy && strategy.cancelUndispatched) ? strategy.cancelUndispatched(toCancel, cancelUndispatched) : cancelUndispatched(toCancel); if (response && (response.error !== undefined || response.result !== undefined)) { + messageQueue.delete(key); + requestTokens.delete(cancelId); response.id = toCancel.id; - handleUnqueuedMessage(message, () => { - // The hook can call next later or not at all. Only drop the - // request if it is still queued then, otherwise cancel it. - if (messageQueue.get(key) !== toCancel) { - requestTokens.get(cancelId)?.cancel(); - traceReceivedNotification(message); - return; - } - messageQueue.delete(key); - requestTokens.delete(cancelId); - traceSendingResponse(response, message.method, Date.now()); - return writeMessage(response).catch(() => logger.error(`Sending response for canceled message failed.`)); - }); + traceSendingResponse(response, message.method, Date.now()); + writeMessage(response).catch(() => logger.error(`Sending response for canceled message failed.`)); return; } } const cancellationToken = requestTokens.get(cancelId); // The request is already running. Cancel the token if (cancellationToken !== undefined) { - handleUnqueuedMessage(message, () => { - cancellationToken.cancel(); - traceReceivedNotification(message); - }); + cancellationToken.cancel(); + traceReceivedNotification(message); return; } else { // Remember the cancel but still queue the message to @@ -882,7 +888,7 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } finally { triggerMessageQueue(); } - }; + } async function handleRequest(requestMessage: RequestMessage): Promise { if (isDisposed()) { diff --git a/jsonrpc/src/node/test/connection.test.ts b/jsonrpc/src/node/test/connection.test.ts index 2b9b27a70..219ccaf3d 100644 --- a/jsonrpc/src/node/test/connection.test.ts +++ b/jsonrpc/src/node/test/connection.test.ts @@ -1127,6 +1127,51 @@ suite('Connection', () => { assert.strictEqual(await second, 2); }); + test('Message strategy can ignore the cancellation of a queued request without cancelUndispatched', async () => { + const type = new RequestType('test/handleSingleRequest'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger, { + maxParallelism: 1, + messageStrategy: { + handleIncomingMessage(message, next) { + // The cancellation of the queued request (id 1) is dropped. + if (hostConnection.Message.isNotification(message) && message.method === '$/cancelRequest' && (message.params as { id: number }).id === 1) { + return; + } + return next(message); + } + } + }); + let started!: () => void; + const firstStarted = new Promise(resolve => started = resolve); + server.onRequest(type, (param, token) => { + if (param === 1) { + started(); + return new Promise(resolve => token.onCancellationRequested(() => resolve(true))); + } + return token.isCancellationRequested; + }); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger); + client.listen(); + + // The second request waits in the queue while the first one runs. The + // first one ends on its own cancellation, which is read after the + // dropped one. + const firstSource = new CancellationTokenSource(); + const first = client.sendRequest(type, 1, firstSource.token); + await firstStarted; + const secondSource = new CancellationTokenSource(); + const second = client.sendRequest(type, 2, secondSource.token); + secondSource.cancel(); + firstSource.cancel(); + assert.strictEqual(await first, true); + assert.strictEqual(await second, false); + }); + test('Message strategy errors on messages that skip the queue are logged', () => { const proc = new EventEmitter(); const errors: string[] = []; From 8575cc1fbdb9715e836f2e9b0015b868628c38d6 Mon Sep 17 00:00:00 2001 From: kwy404 Date: Thu, 5 Nov 2026 14:09:01 -0300 Subject: [PATCH 5/7] Handle hooks that misuse next and deprecate handleMessage next in handleOutgoingMessage now only works once. If the hook settles without calling it, writeMessage rejects, so a request fails with MessageWriteError instead of waiting forever and a notification rejects. Queued messages no longer reach the hook or the writer after dispose. handleMessage is deprecated in favor of handleIncomingMessage, the requirement to call next for $/cancelRequest is documented, and the check for handleIncomingMessage moved to an incomingHook helper. --- README.md | 6 + jsonrpc/README.md | 6 + jsonrpc/src/common/connection.ts | 62 ++++++++-- jsonrpc/src/node/test/connection.test.ts | 138 ++++++++++++++++++++++- 4 files changed, 202 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 5ea87fcb3..5729b65c7 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,12 @@ See [CONTRIBUTING.md](./CONTRIBUTING.md) for the repository map, focused build a ## History +## Next + +### Compatibility + +- JSON-RPC: `handleMessage` is now optional on `MessageStrategy` and deprecated in favor of `handleIncomingMessage`. Code that calls `strategy.handleMessage(...)` directly needs a check for `undefined` when `strictNullChecks` is on. + ## 3.18.5 Protocol, 10.0.0 JSON-RPC, 11.0.0 Client and 11.0.0 Server. [Fixed issues](https://github.com/microsoft/vscode-languageserver-node/issues?q=is%3Aissue%20is%3Aclosed%20reason%3Acompleted%20milestone%3A%223.18.5%22%20sort%3Aupdated-desc) | [Merged pull requests](https://github.com/microsoft/vscode-languageserver-node/pulls?q=is%3Apr%20is%3Amerged%20milestone%3A%223.18.5%22%20sort%3Aupdated-desc) diff --git a/jsonrpc/README.md b/jsonrpc/README.md index ba6160b0a..77e05cd4b 100644 --- a/jsonrpc/README.md +++ b/jsonrpc/README.md @@ -66,6 +66,12 @@ let connection = rpc.createMessageConnection(reader, writer, undefined, { }); ``` +A `handleIncomingMessage` hook must call `next` for `$/cancelRequest`, or the request is not canceled. + +A `handleOutgoingMessage` hook must call `next` exactly once before its promise settles. If it does not call `next`, the message is not written and sending it fails: a request is rejected with `ErrorCodes.MessageWriteError` and the promise of a notification is rejected. Any other call to `next` logs a warning and writes nothing. + +Properties added at the top level of a message reach the writer unchanged. `Message` does not declare them, which is why the example casts the object passed to `next`. + # History For the history please see the [main repository](https://github.com/Microsoft/vscode-languageserver-node/blob/master/README.md) diff --git a/jsonrpc/src/common/connection.ts b/jsonrpc/src/common/connection.ts index 14b496560..927cd0f26 100644 --- a/jsonrpc/src/common/connection.ts +++ b/jsonrpc/src/common/connection.ts @@ -478,6 +478,8 @@ export interface MessageStrategy { * Handles a received message before it is dispatched. Not called if * `handleIncomingMessage` is provided. * + * @deprecated Use `handleIncomingMessage` instead. + * * @param message The received message. * @param next Dispatches the message. */ @@ -487,8 +489,10 @@ export interface MessageStrategy { * Handles a received message before it is dispatched. If provided, it is * used instead of `handleMessage`. Unlike `handleMessage`, it also gets the * messages that skip the message queue: responses when `maxParallelism` is - * set, and `$/cancelRequest` notifications as soon as they are read. A - * request is only canceled once `next` is called for its `$/cancelRequest`. + * set, and `$/cancelRequest` notifications as soon as they are read. + * + * You must call `next` for `$/cancelRequest`, or the request is not + * canceled. * * @param message The received message. * @param next Dispatches the message. @@ -501,6 +505,16 @@ export interface MessageStrategy { * order: the next one is passed when the promise returned for the previous * one settles. * + * Properties added at the top level of a message reach the writer + * unchanged. `Message` does not declare them, so the object passed to + * `next` has to be cast to `Message`. + * + * You must call `next` exactly once before the returned promise settles. + * If it is not called, the message is not written and sending it fails: a + * request is rejected with `ErrorCodes.MessageWriteError` and the promise + * of a notification is rejected. Any other call to `next` logs a warning + * and writes nothing. + * * @param message The message to send. * @param next Writes the message. Pass it the message to write. */ @@ -765,9 +779,10 @@ export function createMessageConnection(messageReader: MessageReader, messageWri try { inFlight++; const messageStrategy = options?.messageStrategy; - if (messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage)) { + const handleIncoming = incomingHook(); + if (handleIncoming !== undefined) { // A queued cancellation already went through the hook when it was read. - result = isCancelNotification(message) ? handleMessage(message) : messageStrategy.handleIncomingMessage(message, handleMessage); + result = isCancelNotification(message) ? handleMessage(message) : handleIncoming(message, handleMessage); } else if (messageStrategy !== undefined && Is.func(messageStrategy.handleMessage)) { result = messageStrategy.handleMessage(message, handleMessage); } else { @@ -808,14 +823,22 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } } + function incomingHook(): MessageStrategy['handleIncomingMessage'] { + const messageStrategy = options?.messageStrategy; + if (messageStrategy === undefined || !Is.func(messageStrategy.handleIncomingMessage)) { + return undefined; + } + return (message, next) => messageStrategy.handleIncomingMessage!(message, next); + } + // Messages that skip the queue go through `handleIncomingMessage`. // `handleMessage` keeps getting queued messages only. Errors are logged, // since the reader can call us from an event handler. function handleUnqueuedMessage(message: Message, handle: (message: Message) => NotificationResult): void { - const messageStrategy = options?.messageStrategy; + const handleIncoming = incomingHook(); const logError = (error: any) => logger.error(`Processing message failed: ${error}`); try { - const result = messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage) ? messageStrategy.handleIncomingMessage(message, handle) : handle(message); + const result = handleIncoming !== undefined ? handleIncoming(message, handle) : handle(message); if (result instanceof Promise) { result.catch(logError); } @@ -832,16 +855,37 @@ export function createMessageConnection(messageReader: MessageReader, messageWri } // Pass one message at a time to the hook. Otherwise a hook that awaits // before calling next lets a later message reach the writer first. - const result = lastOutgoingMessage.then(() => messageStrategy.handleOutgoingMessage!(message, (message) => messageWriter.write(message))); + const result = lastOutgoingMessage.then(async () => { + if (isDisposed()) { + throw new ConnectionError(ConnectionErrors.Disposed, 'Connection is disposed.'); + } + let written: Promise | undefined; + let settled = false; + const next = (outgoing: Message): Promise => { + if (written !== undefined || settled) { + logger.warn(`handleOutgoingMessage must call next exactly once before it settles.`); + return written ?? Promise.resolve(); + } + written = messageWriter.write(outgoing); + return written; + }; + try { + await messageStrategy.handleOutgoingMessage!(message, next); + } finally { + settled = true; + } + if (written === undefined) { + throw new Error(`handleOutgoingMessage did not call next, so the message was not written.`); + } + }); lastOutgoingMessage = result.catch(() => undefined); return result; } const callback: DataCallback = (message) => { - const messageStrategy = options?.messageStrategy; // The hook gets cancellations as soon as they are read and can drop them, // so nothing is canceled or remembered before it calls next. - if (isCancelNotification(message) && messageStrategy !== undefined && Is.func(messageStrategy.handleIncomingMessage)) { + if (isCancelNotification(message) && incomingHook() !== undefined) { handleUnqueuedMessage(message, receiveMessage); } else { receiveMessage(message); diff --git a/jsonrpc/src/node/test/connection.test.ts b/jsonrpc/src/node/test/connection.test.ts index 219ccaf3d..ffd8d1885 100644 --- a/jsonrpc/src/node/test/connection.test.ts +++ b/jsonrpc/src/node/test/connection.test.ts @@ -12,7 +12,7 @@ import { inherits } from 'util'; import { AsyncLocalStorage } from 'async_hooks'; import { EventEmitter } from 'events'; -import { CancellationTokenSource, RequestType, RequestType3, ResponseError, NotificationType, NotificationType2, ErrorCodes, Trace, Tracer } from '../main'; +import { CancellationTokenSource, RequestType, RequestType0, RequestType3, ResponseError, NotificationType, NotificationType2, ErrorCodes, Trace, Tracer } from '../main'; import * as hostConnection from '../main'; import { getCustomCancellationStrategy } from './customCancellationStrategy'; @@ -1007,6 +1007,142 @@ suite('Connection', () => { assert.deepStrictEqual(received, [didOpen.method, didChange.method]); }); + test('Message strategy fails a request when handleOutgoingMessage does not call next', async () => { + const type = new RequestType0('test/request'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { + messageStrategy: { + async handleOutgoingMessage() { + } + } + }); + client.listen(); + + await assert.rejects(client.sendRequest(type), (error: ResponseError) => error.code === ErrorCodes.MessageWriteError); + }); + + test('Message strategy fails a notification when handleOutgoingMessage does not call next', async () => { + const type = new NotificationType('test/notification'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { + messageStrategy: { + async handleOutgoingMessage() { + } + } + }); + client.listen(); + + await assert.rejects(client.sendNotification(type, 1)); + }); + + test('Message strategy writes a message once when handleOutgoingMessage calls next twice', async () => { + const first = new NotificationType('test/first'); + const second = new NotificationType('test/second'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const received: string[] = []; + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger); + const secondReceived = new Promise((resolve) => { + server.onNotification((method) => { + received.push(method); + if (method === second.method) { + resolve(); + } + }); + }); + server.listen(); + + const warnings: string[] = []; + const logger: hostConnection.Logger = { ...hostConnection.NullLogger, warn: (message) => warnings.push(message) }; + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, logger, { + messageStrategy: { + async handleOutgoingMessage(message, next) { + await next(message); + if (hostConnection.Message.isNotification(message) && message.method === first.method) { + await next(message); + } + } + } + }); + client.listen(); + + void client.sendNotification(first, 1); + void client.sendNotification(second, 2); + await secondReceived; + assert.deepStrictEqual(received, [first.method, second.method]); + assert.strictEqual(warnings.length, 1); + }); + + for (const failure of ['throws', 'rejects']) { + test(`Message strategy keeps writing after handleOutgoingMessage ${failure}`, async () => { + const failing = new NotificationType('test/failing'); + const after = new NotificationType('test/after'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + const server = hostConnection.createMessageConnection(duplexStream2, duplexStream1, hostConnection.NullLogger); + const received = new Promise((resolve) => { + server.onNotification((method) => resolve(method)); + }); + server.listen(); + + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { + messageStrategy: { + handleOutgoingMessage(message, next) { + if (hostConnection.Message.isNotification(message) && message.method === failing.method) { + if (failure === 'throws') { + throw new Error('hook failed'); + } + return Promise.reject(new Error('hook failed')); + } + return next(message); + } + } + }); + client.listen(); + + await assert.rejects(client.sendNotification(failing, 1), /hook failed/); + await client.sendNotification(after, 2); + assert.strictEqual(await received, after.method); + }); + } + + test('Message strategy does not handle queued outgoing messages after dispose', async () => { + const first = new NotificationType('test/first'); + const second = new NotificationType('test/second'); + const duplexStream1 = new TestDuplex('ds1'); + const duplexStream2 = new TestDuplex('ds2'); + + let release!: () => void; + const released = new Promise((resolve) => release = resolve); + const handled: string[] = []; + const client = hostConnection.createMessageConnection(duplexStream1, duplexStream2, hostConnection.NullLogger, { + messageStrategy: { + async handleOutgoingMessage(message, next) { + handled.push((message as hostConnection.NotificationMessage).method); + await released; + return next(message); + } + } + }); + client.listen(); + + const firstSent = client.sendNotification(first, 1); + const secondSent = client.sendNotification(second, 2); + await new Promise(resolve => setTimeout(resolve, 0)); + client.dispose(); + release(); + + await firstSent.catch(() => undefined); + await assert.rejects(secondSent, (error: hostConnection.ConnectionError) => error.code === hostConnection.ConnectionErrors.Disposed); + assert.deepStrictEqual(handled, [first.method]); + }); + test('Message strategy handles responses that skip the queue', async () => { const type = new RequestType('test/handleSingleRequest'); const duplexStream1 = new TestDuplex('ds1'); From fbf0378a6d45c08965585920301e821bd1a0ed7a Mon Sep 17 00:00:00 2001 From: kwy404 Date: Thu, 5 Nov 2026 14:26:47 -0300 Subject: [PATCH 6/7] Report a failed write when the hook does not return next writeMessage only waited for the hook, so with a hook that called next without returning its promise a failed write was lost and the request waited forever. --- jsonrpc/src/common/connection.ts | 3 +++ jsonrpc/src/node/test/connection.test.ts | 23 +++++++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/jsonrpc/src/common/connection.ts b/jsonrpc/src/common/connection.ts index 927cd0f26..bea456235 100644 --- a/jsonrpc/src/common/connection.ts +++ b/jsonrpc/src/common/connection.ts @@ -877,6 +877,9 @@ export function createMessageConnection(messageReader: MessageReader, messageWri if (written === undefined) { throw new Error(`handleOutgoingMessage did not call next, so the message was not written.`); } + // A hook may not return the promise of next, so a failed write has to + // be reported here or the request would wait forever. + await written; }); lastOutgoingMessage = result.catch(() => undefined); return result; diff --git a/jsonrpc/src/node/test/connection.test.ts b/jsonrpc/src/node/test/connection.test.ts index ffd8d1885..8e0898c89 100644 --- a/jsonrpc/src/node/test/connection.test.ts +++ b/jsonrpc/src/node/test/connection.test.ts @@ -1039,6 +1039,29 @@ suite('Connection', () => { await assert.rejects(client.sendNotification(type, 1)); }); + test('Message strategy fails a request when a write fails and the hook does not return next', async () => { + class FailingWriter extends hostConnection.AbstractMessageWriter implements hostConnection.MessageWriter { + public write(): Promise { + return Promise.reject(new Error('write failed')); + } + public end(): void { + } + } + const type = new RequestType0('test/request'); + const duplexStream = new TestDuplex('ds1'); + + const client = hostConnection.createMessageConnection(new hostConnection.StreamMessageReader(duplexStream), new FailingWriter(), hostConnection.NullLogger, { + messageStrategy: { + async handleOutgoingMessage(message, next) { + void next(message); + } + } + }); + client.listen(); + + await assert.rejects(client.sendRequest(type), (error: ResponseError) => error.code === ErrorCodes.MessageWriteError); + }); + test('Message strategy writes a message once when handleOutgoingMessage calls next twice', async () => { const first = new NotificationType('test/first'); const second = new NotificationType('test/second'); From 8a7946e9248c6daec0d5eba85d5da306aac3b0f5 Mon Sep 17 00:00:00 2001 From: kwy404 Date: Fri, 6 Nov 2026 21:03:37 -0300 Subject: [PATCH 7/7] Do not leave a failed write unhandled while the hook runs A hook that calls next without awaiting it and is still running when the write fails left the rejection without a handler, which can end the Node.js process even though the caller handles the send failure. Mark the promise as handled when it is created. The failure is still reported when the hook settles. --- jsonrpc/src/common/connection.ts | 3 +++ jsonrpc/src/node/test/connection.test.ts | 34 ++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/jsonrpc/src/common/connection.ts b/jsonrpc/src/common/connection.ts index bea456235..e7d0b495f 100644 --- a/jsonrpc/src/common/connection.ts +++ b/jsonrpc/src/common/connection.ts @@ -867,6 +867,9 @@ export function createMessageConnection(messageReader: MessageReader, messageWri return written ?? Promise.resolve(); } written = messageWriter.write(outgoing); + // The rejection is reported once the hook settles. Until then the + // hook may not be listening, which would be an unhandled rejection. + written.catch(() => undefined); return written; }; try { diff --git a/jsonrpc/src/node/test/connection.test.ts b/jsonrpc/src/node/test/connection.test.ts index 8e0898c89..d017a4414 100644 --- a/jsonrpc/src/node/test/connection.test.ts +++ b/jsonrpc/src/node/test/connection.test.ts @@ -1062,6 +1062,40 @@ suite('Connection', () => { await assert.rejects(client.sendRequest(type), (error: ResponseError) => error.code === ErrorCodes.MessageWriteError); }); + test('Message strategy does not leave a failed write unhandled while the hook runs', async () => { + class FailingWriter extends hostConnection.AbstractMessageWriter implements hostConnection.MessageWriter { + public write(): Promise { + return Promise.reject(new Error('write failed')); + } + public end(): void { + } + } + const type = new RequestType0('test/request'); + const duplexStream = new TestDuplex('ds1'); + + const unhandled: unknown[] = []; + const listeners = process.listeners('unhandledRejection'); + process.removeAllListeners('unhandledRejection'); + process.on('unhandledRejection', (reason) => unhandled.push(reason)); + try { + const client = hostConnection.createMessageConnection(new hostConnection.StreamMessageReader(duplexStream), new FailingWriter(), hostConnection.NullLogger, { + messageStrategy: { + async handleOutgoingMessage(message, next) { + void next(message); + await new Promise(resolve => setTimeout(resolve, 10)); + } + } + }); + client.listen(); + + await assert.rejects(client.sendRequest(type), (error: ResponseError) => error.code === ErrorCodes.MessageWriteError); + assert.deepStrictEqual(unhandled, []); + } finally { + process.removeAllListeners('unhandledRejection'); + listeners.forEach(listener => process.on('unhandledRejection', listener)); + } + }); + test('Message strategy writes a message once when handleOutgoingMessage calls next twice', async () => { const first = new NotificationType('test/first'); const second = new NotificationType('test/second');