From 57dceb6f7f0982746cb5edcf66ab94a0776cc8e9 Mon Sep 17 00:00:00 2001 From: wsj Date: Sun, 1 Nov 2026 19:01:27 +0800 Subject: [PATCH 1/3] Reject pending async API requests on connection close --- packages/typescript/src/api/async/client.ts | 45 +- packages/typescript/src/api/async/jsonrpc.ts | 149 ++++- .../typescript/test/async/disconnect.test.ts | 526 ++++++++++++++++++ 3 files changed, 701 insertions(+), 19 deletions(-) create mode 100644 packages/typescript/test/async/disconnect.test.ts diff --git a/packages/typescript/src/api/async/client.ts b/packages/typescript/src/api/async/client.ts index bb5adf7030aa3..09feff148c869 100644 --- a/packages/typescript/src/api/async/client.ts +++ b/packages/typescript/src/api/async/client.ts @@ -70,6 +70,7 @@ class RpcClient { private timing: TimingCollector | undefined; private timingDump: TimingDump | undefined; private closing: Promise | undefined; + private draining: Promise | undefined; private batchedRequests: { method: APIRequest["method"]; params: APIRequest["params"]; resolve: (value: unknown) => void; reject: (reason?: any) => void; }[] = []; private nextBatch: NodeJS.Immediate | "manual" | undefined; @@ -129,9 +130,9 @@ class RpcClient { resolve(); }); - this.connection = this.jsonrpc!.createStreamConnection(this.process.stdout!, this.process.stdin!); - this.registerFSCallbacks(this.connection, options.fs, fsConfiguration); - this.connection.listen(); + const connection = this.setConnection(this.jsonrpc!.createStreamConnection(this.process.stdout!, this.process.stdin!)); + this.registerFSCallbacks(connection, options.fs, fsConfiguration); + connection.listen(); }); } @@ -140,8 +141,8 @@ class RpcClient { return new Promise((resolve, reject) => { this.socket = createConnection(options.pipe, () => { - this.connection = this.jsonrpc!.createStreamConnection(this.socket!, this.socket!); - this.connection.listen(); + const connection = this.setConnection(this.jsonrpc!.createStreamConnection(this.socket!, this.socket!)); + connection.listen(); this.connected = true; resolve(); }); @@ -152,6 +153,23 @@ class RpcClient { }); } + private setConnection({ connection, drain }: ReturnType): JsonRpc.MessageConnection { + this.connection = connection; + this.draining = undefined; + connection.onClose(() => { + // The JSON-RPC connection emits close without rejecting pending requests. + this.connectionClosing(connection); + const draining = drain().then(() => connection.dispose()); + if (this.connection === connection) this.draining = draining; + }); + return connection; + } + + private connectionClosing(connection: JsonRpc.MessageConnection): void { + // A failed spawn may close its streams after a later connection is established. + if (this.connected && this.connection === connection) this.rejectQueuedRequests(); + } + private registerFSCallbacks( connection: JsonRpc.MessageConnection, fs: FileSystemCallbacks | undefined, @@ -423,7 +441,7 @@ class RpcClient { return this.closing ??= this.closeWorker(); } - private async closeWorker(): Promise { + private rejectQueuedRequests(): void { this.closed = true; if (this.nextBatch && this.nextBatch !== "manual") { clearImmediate(this.nextBatch); @@ -434,29 +452,28 @@ class RpcClient { for (const { reject } of requests) { reject(new Error("Client is closed")); } + } + + private async closeWorker(): Promise { + this.rejectQueuedRequests(); await this.connecting?.catch(() => {}); // if connection is still in-progress, wait for it to finish before closing the connection try { if (this.timingDump && this.timing) { - const local = this.timing.getInfo(); let server: ServerTimingInfo | undefined; let serverTimingError: string | undefined; - const connection = this.connection; - // Closing alone does not reject pending JSON-RPC requests. - const onClose = connection?.onClose(() => connection.dispose()); try { if (this.connected) server = await this.fetchServerTiming(); } catch (error) { serverTimingError = String(error); } - finally { - onClose?.dispose(); - } - this.timingDump.write(local, server, serverTimingError); + if (this.draining) await this.draining; + this.timingDump.write(this.timing.getInfo(), server, serverTimingError); } } finally { if (this.connection) { + if (this.draining) await this.draining; this.connection.dispose(); this.connection = undefined; } diff --git a/packages/typescript/src/api/async/jsonrpc.ts b/packages/typescript/src/api/async/jsonrpc.ts index fec7364099f04..7213b30193118 100644 --- a/packages/typescript/src/api/async/jsonrpc.ts +++ b/packages/typescript/src/api/async/jsonrpc.ts @@ -6,6 +6,7 @@ import { AbstractMessageBuffer, createMessageConnection, Disposable, + Message, type MessageConnection, NullLogger, RAL, @@ -97,16 +98,74 @@ function installRuntime(): void { } /** Creates a JSON-RPC connection over Node.js streams. */ -export function createStreamConnection(readable: EventSource, writable: WritableStream): MessageConnection { +export function createStreamConnection(readable: EventSource, writable: WritableStream): { connection: MessageConnection; drain(): Promise; } { installRuntime(); + const frames = new IncomingFrames(); + const responses = new Map(); + let closing = false; + let resolveDrain: (() => void) | undefined; + + function finishDrain(): void { + if (frames.pending === 0 && responses.size === 0 && resolveDrain) { + resolveDrain(); + resolveDrain = undefined; + } + } + + function onClose(stream: EventSource, listener: () => void): Disposable { + return listen(stream, "close", () => { + // A write-side close can precede EOF on the read side. Only frames + // received before close belong to this drain; do not wait for later data. + closing = true; + listener(); + }); + } + + const decoder = RAL().applicationJson.decoder; const reader = new ReadableStreamMessageReader({ - onClose: listener => listen(readable, "close", listener), + onClose: listener => onClose(readable, listener), onError: listener => listen(readable, "error", listener), onEnd: listener => listen(readable, "end", listener), - onData: listener => listen(readable, "data", listener), + onData: listener => + listen(readable, "data", (data: Uint8Array | string) => { + if (closing) return; + const bytes = typeof data === "string" ? Buffer.from(data, "utf8") : data; + frames.append(bytes); + listener(bytes); + }), + }, { + contentTypeDecoder: { + name: decoder.name, + async decode(data, options) { + try { + return await decoder.decode(data, options); + } + catch (error) { + frames.finish(); + finishDrain(); + throw error; + } + }, + }, }); + // This connection does not consume partial-message events. In particular, + // avoid leaving their repeating timer alive after a truncated frame closes. + reader.partialMessageTimeout = 0; + const listenForMessages = reader.listen.bind(reader); + reader.listen = callback => + listenForMessages(message => { + const key = responseKey(message); + if (key !== undefined) responses.set(key, message); + try { + callback(message); + } + finally { + frames.finish(); + finishDrain(); + } + }); const writer = new WriteableStreamMessageWriter({ - onClose: listener => listen(writable, "close", listener), + onClose: listener => onClose(writable, listener), onError: listener => listen(writable, "error", listener), onEnd: listener => listen(writable, "end", listener), write: (data: Uint8Array | string, encoding?: RAL.MessageBufferEncoding) => @@ -121,5 +180,85 @@ export function createStreamConnection(readable: EventSource, writable: Writable }), end: () => writable.end(), }); - return createMessageConnection(reader, writer, NullLogger); + const connection = createMessageConnection(reader, writer, NullLogger, { + messageStrategy: { + handleMessage(message, next) { + try { + return next(message); + } + finally { + const key = responseKey(message); + if (key !== undefined && responses.get(key) === message) responses.delete(key); + finishDrain(); + } + }, + }, + }); + return { + connection, + drain: () => + frames.pending === 0 && responses.size === 0 + ? Promise.resolve() + : new Promise(resolve => { + resolveDrain = resolve; + }), + }; +} + +function responseKey(message: Message): string | undefined { + // Match the vendored connection's request-first classification and its + // response queue key, which stringifies both numeric and string IDs. + return !Message.isRequest(message) && Message.isResponse(message) && message.id !== null + ? String(message.id) + : undefined; +} + +/** Counts complete Content-Length frames without retaining or copying their bodies. */ +class IncomingFrames { + private header = ""; + private remaining: number | undefined; + private valid = true; + pending = 0; + + append(data: Uint8Array): void { + if (!this.valid) return; + let offset = 0; + while (offset < data.length) { + if (this.remaining !== undefined) { + const consumed = Math.min(this.remaining, data.length - offset); + this.remaining -= consumed; + offset += consumed; + if (this.remaining === 0) { + this.pending++; + this.remaining = undefined; + } + continue; + } + this.header += String.fromCharCode(data[offset++] & 0x7f); + if (!this.header.endsWith("\r\n\r\n")) continue; + let contentLength: string | undefined; + for (const line of this.header.slice(0, -4).split("\r\n")) { + const separator = line.indexOf(":"); + if (separator < 0) { + this.valid = false; + return; + } + if (line.slice(0, separator).toLowerCase() === "content-length") contentLength = line.slice(separator + 1).trim(); + } + const length = contentLength ? parseInt(contentLength) : NaN; + if (!Number.isSafeInteger(length) || length < 0) { + this.valid = false; + return; + } + this.header = ""; + if (length === 0) this.pending++; + else this.remaining = length; + } + } + + finish(): void { + // After an invalid header, the reader may recover on a later data event + // even though frame tracking has stopped. + if (this.pending > 0) this.pending--; + } } diff --git a/packages/typescript/test/async/disconnect.test.ts b/packages/typescript/test/async/disconnect.test.ts new file mode 100644 index 0000000000000..bc5ab8dd8db06 --- /dev/null +++ b/packages/typescript/test/async/disconnect.test.ts @@ -0,0 +1,526 @@ +import assert from "node:assert/strict"; +import { type ChildProcess } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { + EventEmitter, + once, +} from "node:events"; +import { + mkdtempSync, + readdirSync, + readFileSync, + rmSync, +} from "node:fs"; +import { + createServer, + type Socket, +} from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { PassThrough } from "node:stream"; +import { + test, + type TestContext, +} from "node:test"; +import { + type Client, + createClient, +} from "../../src/api/async/client.ts"; +import { + createStreamConnection, + type MessageConnection, + RequestType, +} from "../../src/api/async/jsonrpc.ts"; +import { + createMessageConnection, + Message, + SocketMessageReader, + SocketMessageWriter, + StreamMessageReader, + StreamMessageWriter, +} from "../../vendor/vscode-jsonrpc/lib/node/main.js"; + +type TransportKind = "socket" | "spawn"; + +interface Peer { + client: Client; + requests: EventEmitter; + disconnect(): void; + replyAndDisconnect?: () => void; +} + +const parsedCommandLine = { options: {}, fileNames: [], errors: [] }; + +async function createPeer(t: TestContext, kind: TransportKind, holdInitialize = false, timingDirectory?: string): Promise { + const previousDirectory = process.env.TS_API_TIMING_DIR; + if (timingDirectory === undefined) delete process.env.TS_API_TIMING_DIR; + else process.env.TS_API_TIMING_DIR = timingDirectory; + t.after(() => { + if (previousDirectory === undefined) delete process.env.TS_API_TIMING_DIR; + else process.env.TS_API_TIMING_DIR = previousDirectory; + }); + const requests = new EventEmitter(); + if (kind === "socket") { + const pipe = process.platform === "win32" + ? `\\\\.\\pipe\\ts-api-disconnect-${randomUUID()}` + : join(tmpdir(), `ts-api-disconnect-${randomUUID()}.sock`); + let socket: Socket | undefined; + const queuedIds: (number | string)[] = []; + const server = createServer(value => { + socket = value; + const connection = createMessageConnection(new SocketMessageReader(value), new SocketMessageWriter(value), undefined, { + messageStrategy: { + handleMessage(message, next) { + if ( + Message.isRequest(message) && message.id !== null && message.method === "parseCommandLine" && + (message.params as { commandLine: string[]; }).commandLine[0] === "queue" + ) { + queuedIds.push(message.id); + requests.emit("request", "queue"); + return; + } + if ( + Message.isRequest(message) && message.method === "parseCommandLine" && + (message.params as { commandLine: string[]; }).commandLine[0] === "replyAndClose" + ) { + const response = JSON.stringify({ jsonrpc: "2.0", id: message.id, result: parsedCommandLine }); + value.end(`Content-Length: ${Buffer.byteLength(response)}\r\n\r\n${response}`); + return; + } + return next(message); + }, + }, + }); + connection.onRequest("initialize", () => { + if (holdInitialize) { + requests.emit("request", "initialize"); + return new Promise(() => {}); + } + return { currentDirectory: process.cwd(), caseSensitivity: 1 }; + }); + connection.onRequest("parseCommandLine", (params: { commandLine: string[]; }) => { + if (params.commandLine[0] === "hold") { + requests.emit("request", "parseCommandLine"); + return new Promise(() => {}); + } + return parsedCommandLine; + }); + connection.onRequest("batchRequests", () => { + requests.emit("request", "batchRequests"); + return new Promise(() => {}); + }); + connection.onRequest("getServerTiming", () => { + requests.emit("request", "getServerTiming"); + return new Promise(() => {}); + }); + connection.listen(); + }); + server.listen(pipe); + await once(server, "listening"); + const client = createClient({ pipe }); + t.after(async () => { + await client.close(); + socket?.destroy(); + await new Promise(resolve => server.close(() => resolve())); + }); + return { + client, + requests, + disconnect: () => socket!.destroy(), + replyAndDisconnect: () => socket!.end(Buffer.concat(queuedIds.map(id => responseFrame(id, parsedCommandLine)))), + }; + } + + // The client still uses a real spawned process and stdio streams. Replace only + // the compiler executable with a small Node JSON-RPC peer for this test. + const childProcess = process.getBuiltinModule("node:child_process"); + const spawn = childProcess.spawn; + let child: ChildProcess | undefined; + const script = ` + import { createMessageConnection, Message, StreamMessageReader, StreamMessageWriter } from ${JSON.stringify(new URL("../../vendor/vscode-jsonrpc/lib/node/main.js", import.meta.url).href)}; + const connection = createMessageConnection(new StreamMessageReader(process.stdin), new StreamMessageWriter(process.stdout), undefined, { + messageStrategy: { + handleMessage(message, next) { + if (Message.isRequest(message) && message.method === "parseCommandLine" && message.params.commandLine[0] === "replyAndClose") { + const response = JSON.stringify({ jsonrpc: "2.0", id: message.id, result: { options: {}, fileNames: [], errors: [] } }); + process.stdout.end("Content-Length: " + Buffer.byteLength(response) + "\\r\\n\\r\\n" + response, () => process.exit(0)); + return; + } + return next(message); + }, + }, + }); + connection.onRequest("initialize", () => { + if (${holdInitialize}) { + process.send("initialize"); + return new Promise(() => {}); + } + return { currentDirectory: process.cwd(), caseSensitivity: 1 }; + }); + connection.onRequest("parseCommandLine", params => { + if (params.commandLine[0] === "hold") { + process.send("parseCommandLine"); + return new Promise(() => {}); + } + return { options: {}, fileNames: [], errors: [] }; + }); + connection.onRequest("batchRequests", () => { + process.send("batchRequests"); + return new Promise(() => {}); + }); + connection.listen(); + process.on("message", () => process.exit(0)); + process.stdin.on("end", () => process.exit(0)); + `; + t.mock.method(childProcess, "spawn", () => { + child = spawn(process.execPath, ["--input-type=module", "--eval", script], { stdio: ["pipe", "pipe", "inherit", "ipc"] }); + child.on("message", message => requests.emit("request", message)); + return child; + }); + const client = createClient({ tsserverPath: process.execPath, useCaseSensitiveFileNames: true }); + t.after(async () => { + await client.close(); + if (child && child.exitCode === null && child.signalCode === null) { + const exited = once(child, "exit"); + child.kill(); + await exited; + } + }); + return { + client, + requests, + disconnect: () => { + child!.send!("disconnect"); + }, + }; +} + +async function assertRejectsPromptly(promise: Promise): Promise { + let timer: NodeJS.Timeout | undefined; + try { + await Promise.race([ + assert.rejects(promise, /closed|disposed/i), + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("Request did not reject after the connection closed")), 3000); + }), + ]); + } + finally { + clearTimeout(timer); + } +} + +for (const kind of ["socket", "spawn"] as const) { + test(`${kind}: receives a response before a graceful disconnect`, { timeout: 10000 }, async t => { + const { client } = await createPeer(t, kind); + await client.apiRequest("initialize", null); + assert.deepEqual(await client.apiRequest("parseCommandLine", { commandLine: ["replyAndClose"] }), parsedCommandLine); + }); + + test(`${kind}: normal requests work and explicit close rejects an in-flight request`, { timeout: 10000 }, async t => { + const { client, requests } = await createPeer(t, kind); + await client.apiRequest("initialize", null); + assert.deepEqual(await client.apiRequest("parseCommandLine", { commandLine: [] }), parsedCommandLine); + const received = once(requests, "request"); + const pending = client.apiRequest("parseCommandLine", { commandLine: ["hold"] }); + assert.deepEqual(await received, ["parseCommandLine"]); + const rejected = assertRejectsPromptly(pending); + await client.close(); + await rejected; + await assert.rejects(client.apiRequest("parseCommandLine", { commandLine: [] }), /closed/i); + }); + + test(`${kind}: disconnect rejects all in-flight and batched requests`, { timeout: 10000 }, async t => { + const { client, requests, disconnect } = await createPeer(t, kind); + await client.apiRequest("initialize", null); + const pending: Promise[] = []; + for (let i = 0; i < 2; i++) { + const received = once(requests, "request"); + pending.push(client.apiRequest("parseCommandLine", { commandLine: ["hold"] })); + assert.deepEqual(await received, ["parseCommandLine"]); + } + const received = once(requests, "request"); + const batch = client.batchContext(); + pending.push(client.apiRequest("parseCommandLine", { commandLine: ["hold"] })); + pending.push(client.apiRequest("parseCommandLine", { commandLine: ["hold"] })); + batch[Symbol.dispose](); + assert.deepEqual(await received, ["batchRequests"]); + const rejected = Promise.all(pending.map(assertRejectsPromptly)); + disconnect(); + await rejected; + await assert.rejects(client.apiRequest("parseCommandLine", { commandLine: [] }), /closed|disposed/i); + }); + + test(`${kind}: disconnect rejects requests held by a manual batch`, { timeout: 10000 }, async t => { + const { client, disconnect } = await createPeer(t, kind); + await client.apiRequest("initialize", null); + const batch = client.batchContext(); + const pending = client.apiRequest("parseCommandLine", { commandLine: ["hold"] }); + const rejected = assertRejectsPromptly(pending); + disconnect(); + await rejected; + batch[Symbol.dispose](); + }); + + test(`${kind}: disconnect rejects initialization`, { timeout: 10000 }, async t => { + const { client, requests, disconnect } = await createPeer(t, kind, true); + const received = once(requests, "request"); + const pending = client.apiRequest("initialize", null); + assert.deepEqual(await received, ["initialize"]); + const rejected = assertRejectsPromptly(pending); + disconnect(); + await rejected; + }); +} + +for (const closeTime of ["before", "during", "after"] as const) { + test(`a failed spawn can be retried when its streams close ${closeTime} the retry`, { timeout: 10000 }, async t => { + const childProcess = process.getBuiltinModule("node:child_process"); + const created = new EventEmitter(); + const children: ChildProcess[] = []; + t.mock.method(childProcess, "spawn", () => { + const child = new EventEmitter() as ChildProcess; + child.stdin = new PassThrough(); + child.stdout = new PassThrough(); + children.push(child); + created.emit("child"); + return child; + }); + const client = createClient({ tsserverPath: process.execPath, useCaseSensitiveFileNames: true }); + if (!("connect" in client)) throw new Error("Expected a process client"); + t.after(async () => { + await client.close(); + for (const child of children) { + child.stdin?.destroy(); + child.stdout?.destroy(); + } + }); + + const firstCreated = once(created, "child"); + const first = client.connect(); + await firstCreated; + children[0].emit("error", new Error("temporary failure")); + await assert.rejects(first, /temporary failure/); + async function closeFirstStream(): Promise { + const closed = once(children[0].stdout!, "close"); + children[0].stdout!.destroy(); + await closed; + } + if (closeTime === "before") await closeFirstStream(); + + const secondCreated = once(created, "child"); + const second = client.connect(); + await secondCreated; + if (closeTime === "during") await closeFirstStream(); + const secondChild = children[1]; + const peer = createMessageConnection(new StreamMessageReader(secondChild.stdin as PassThrough), new StreamMessageWriter(secondChild.stdout as PassThrough)); + peer.onRequest("parseCommandLine", () => parsedCommandLine); + peer.listen(); + secondChild.emit("spawn"); + await second; + if (closeTime === "after") await closeFirstStream(); + assert.deepEqual(await client.apiRequest("parseCommandLine", { commandLine: [] }), parsedCommandLine); + peer.dispose(); + }); +} + +function createWirePeer(t: TestContext) { + const input = new PassThrough(); + const output = new PassThrough(); + const received = new EventEmitter(); + const { connection, drain } = createStreamConnection(input, output); + connection.onClose(() => { + void drain().then(() => connection.dispose()); + }); + connection.listen(); + let buffer: Buffer = Buffer.alloc(0); + output.on("data", (chunk: Buffer) => { + buffer = Buffer.concat([buffer, chunk]); + while (true) { + const headerEnd = buffer.indexOf("\r\n\r\n"); + if (headerEnd < 0) return; + const length = Number(/Content-Length: (\d+)/i.exec(buffer.subarray(0, headerEnd).toString())![1]); + const end = headerEnd + 4 + length; + if (buffer.length < end) return; + received.emit("request", JSON.parse(buffer.subarray(headerEnd + 4, end).toString())); + buffer = buffer.subarray(end); + } + }); + t.after(() => { + connection.dispose(); + input.destroy(); + output.destroy(); + }); + return { connection, input, output, received }; +} + +function responseFrame(id: number | string, result: unknown): Buffer { + const body = JSON.stringify({ jsonrpc: "2.0", id, result }); + return Buffer.from(`Content-Length: ${Buffer.byteLength(body)}\r\n\r\n${body}`); +} + +test("a close drains many complete responses and rejects a request with no response", { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const pending: Promise[] = []; + const ids: number[] = []; + for (let i = 0; i < 31; i++) { + const request = once(received, "request"); + pending.push(connection.sendRequest(new RequestType("probe"), {})); + const [message] = await request; + ids.push(message.id); + } + const settled = Promise.allSettled(pending); + input.end(Buffer.concat(ids.slice(0, -1).map(id => responseFrame(id, id)))); + const results = await settled; + assert.deepEqual(results.slice(0, -1), ids.slice(0, -1).map(value => ({ status: "fulfilled", value }))); + assert.equal(results.at(-1)?.status, "rejected"); +}); + +test("a close drains a fragmented complete response", { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + const [message] = await request; + const frame = responseFrame(message.id, "received"); + for (let offset = 0; offset < frame.length; offset += 3) input.write(frame.subarray(offset, offset + 3)); + input.end(); + assert.equal(await pending, "received"); +}); + +for (const [description, invalidFrame] of [["malformed", "Content-Length: 1\r\n\r\n{"], ["empty", "Content-Length: 0\r\n\r\n"]] as const) { + test(`a close drains a valid response after a ${description} JSON frame`, { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + const [message] = await request; + input.end(Buffer.concat([Buffer.from(invalidFrame), responseFrame(message.id, "received")])); + assert.equal(await pending, "received"); + }); +} + +for ( + const [description, data] of [ + ["invalid JSON", "Content-Length: 1\r\n\r\n{"], + ["invalid message", "Content-Length: 4\r\n\r\nnull"], + ["invalid header", "Content-Length: invalid\r\n\r\n{}"], + ["truncated frame", "Content-Length: 40\r\n\r\n{"], + ] as const +) { + test(`a close rejects a request after ${description}`, { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + await request; + const rejected = assertRejectsPromptly(pending); + input.end(data); + await rejected; + }); +} + +test("writer close rejects pending requests without waiting for later input", { timeout: 10000 }, async t => { + const { connection, input, output, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + await request; + const rejected = assertRejectsPromptly(pending); + const closed = once(output, "close"); + output.destroy(); + await closed; + const body = JSON.stringify({ jsonrpc: "2.0", method: "notification" }); + const frame = Buffer.from(`Content-Length: ${Buffer.byteLength(body)}\r\n\r\n${body}`); + for (let i = 0; i < 30; i++) input.write(frame); + await rejected; +}); + +test("a close rejects pending requests after an invalid header followed by another message", { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + await request; + const rejected = assertRejectsPromptly(pending); + const notification = new Promise(resolve => connection.onNotification("notification", resolve)); + input.write("Content-Length: invalid\r\n\r\n"); + const body = JSON.stringify({ jsonrpc: "2.0", method: "notification" }); + input.write(`Content-Length: ${Buffer.byteLength(body)}\r\n\r\n${body}`); + await notification; + input.end(); + await rejected; +}); + +for (const invalidLength of ["invalid", "-1"]) { + test(`a response followed by Content-Length ${invalidLength} does not prevent close`, { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const requests: Promise[] = []; + const ids: number[] = []; + for (let i = 0; i < 2; i++) { + const request = once(received, "request"); + requests.push(connection.sendRequest(new RequestType("probe"), {})); + const [message] = await request; + ids.push(message.id); + } + const settled = Promise.allSettled(requests); + input.write(Buffer.concat([responseFrame(ids[0], "received"), Buffer.from(`Content-Length: ${invalidLength}\r\n\r\n`)])); + const body = JSON.stringify({ jsonrpc: "2.0", method: "notification" }); + input.end(`Content-Length: ${Buffer.byteLength(body)}\r\n\r\n${body}`); + const results = await settled; + assert.deepEqual(results[0], { status: "fulfilled", value: "received" }); + assert.equal(results[1].status, "rejected"); + }); +} + +test("a timing dump can be written after an unexpected disconnect", { timeout: 10000 }, async t => { + const directory = mkdtempSync(join(tmpdir(), "ts-api-disconnect-timing-")); + const { client, requests, disconnect } = await createPeer(t, "socket", false, directory); + t.after(() => rmSync(directory, { recursive: true, force: true })); + await client.apiRequest("initialize", null); + const received = once(requests, "request"); + const pending = client.apiRequest("parseCommandLine", { commandLine: ["hold"] }); + await received; + const rejected = assertRejectsPromptly(pending); + disconnect(); + await rejected; + await client.close(); + const files = readdirSync(directory); + assert.equal(files.length, 1); + assert.match(JSON.parse(readFileSync(join(directory, files[0]), "utf8")).serverTimingError, /closed|disposed/i); +}); + +test("a timing dump can be written when the server disconnects during close", { timeout: 10000 }, async t => { + const directory = mkdtempSync(join(tmpdir(), "ts-api-disconnect-timing-")); + const { client, requests, disconnect } = await createPeer(t, "socket", false, directory); + t.after(() => rmSync(directory, { recursive: true, force: true })); + await client.apiRequest("initialize", null); + const received = once(requests, "request"); + const closing = client.close(); + assert.deepEqual(await received, ["getServerTiming"]); + disconnect(); + await closing; + const files = readdirSync(directory); + assert.equal(files.length, 1); + assert.match(JSON.parse(readFileSync(join(directory, files[0]), "utf8")).serverTimingError, /closed|disposed/i); +}); + +test("a timing dump includes responses drained after disconnect", { timeout: 10000 }, async t => { + const directory = mkdtempSync(join(tmpdir(), "ts-api-disconnect-timing-")); + const { client, requests, replyAndDisconnect } = await createPeer(t, "socket", false, directory); + t.after(() => rmSync(directory, { recursive: true, force: true })); + await client.apiRequest("initialize", null); + const pending: Promise[] = []; + for (let i = 0; i < 30; i++) { + const received = once(requests, "request"); + pending.push(client.apiRequest("parseCommandLine", { commandLine: ["queue"] })); + assert.deepEqual(await received, ["queue"]); + } + assert.ok(replyAndDisconnect); + const connection = (client as unknown as { connection: MessageConnection; }).connection; + const closed = new Promise(resolve => connection.onClose(resolve)); + replyAndDisconnect(); + await closed; + await pending[0]; + await client.close(); + await Promise.all(pending); + const files = readdirSync(directory); + assert.equal(files.length, 1); + const dump = JSON.parse(readFileSync(join(directory, files[0]), "utf8")); + assert.equal(dump.timing.totals.requestCount, 31); +}); From 40eca2a7722527969e7a9127d4d6d1b9d3656154 Mon Sep 17 00:00:00 2001 From: wsj-cmd Date: Tue, 3 Nov 2026 16:54:53 +0800 Subject: [PATCH 2/3] Fix async response draining after malformed headers --- packages/typescript/src/api/async/jsonrpc.ts | 22 +++++--- .../typescript/test/async/disconnect.test.ts | 53 +++++++++++++++++++ 2 files changed, 69 insertions(+), 6 deletions(-) diff --git a/packages/typescript/src/api/async/jsonrpc.ts b/packages/typescript/src/api/async/jsonrpc.ts index 7213b30193118..75af6dc88cc98 100644 --- a/packages/typescript/src/api/async/jsonrpc.ts +++ b/packages/typescript/src/api/async/jsonrpc.ts @@ -213,15 +213,20 @@ function responseKey(message: Message): string | undefined { : undefined; } -/** Counts complete Content-Length frames without retaining or copying their bodies. */ +/** Counts complete Content-Length frames, retaining only bytes deferred after malformed headers. */ class IncomingFrames { private header = ""; private remaining: number | undefined; + private deferred: Uint8Array | undefined; private valid = true; pending = 0; append(data: Uint8Array): void { if (!this.valid) return; + if (this.deferred) { + data = Buffer.concat([this.deferred, data]); + this.deferred = undefined; + } let offset = 0; while (offset < data.length) { if (this.remaining !== undefined) { @@ -236,29 +241,34 @@ class IncomingFrames { } this.header += String.fromCharCode(data[offset++] & 0x7f); if (!this.header.endsWith("\r\n\r\n")) continue; + const header = this.header; + this.header = ""; + // The reader consumes malformed headers, then retries the remaining + // buffered bytes on the next data event, including an empty event. let contentLength: string | undefined; - for (const line of this.header.slice(0, -4).split("\r\n")) { + for (const line of header.slice(0, -4).split("\r\n")) { const separator = line.indexOf(":"); if (separator < 0) { - this.valid = false; + this.deferred = offset < data.length ? data.subarray(offset) : undefined; return; } if (line.slice(0, separator).toLowerCase() === "content-length") contentLength = line.slice(separator + 1).trim(); } const length = contentLength ? parseInt(contentLength) : NaN; + if (Number.isNaN(length)) { + this.deferred = offset < data.length ? data.subarray(offset) : undefined; + return; + } if (!Number.isSafeInteger(length) || length < 0) { this.valid = false; return; } - this.header = ""; if (length === 0) this.pending++; else this.remaining = length; } } finish(): void { - // After an invalid header, the reader may recover on a later data event - // even though frame tracking has stopped. if (this.pending > 0) this.pending--; } } diff --git a/packages/typescript/test/async/disconnect.test.ts b/packages/typescript/test/async/disconnect.test.ts index bc5ab8dd8db06..4373fc0e0ad2c 100644 --- a/packages/typescript/test/async/disconnect.test.ts +++ b/packages/typescript/test/async/disconnect.test.ts @@ -447,6 +447,59 @@ test("a close rejects pending requests after an invalid header followed by anoth await rejected; }); +for ( + const [description, header] of [ + ["invalid length", "Content-Length: invalid\r\n\r\n"], + ["missing length", "Content-Type: application/json\r\n\r\n"], + ["missing separator", "Content-Length 40\r\n\r\n"], + ] as const +) { + for (const buffered of [false, true]) { + test(`a close drains a recovered response after ${description} (${buffered ? "buffered" : "separate"})`, { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + const [message] = await request; + const frame = responseFrame(message.id, "received"); + if (buffered) { + input.write(Buffer.concat([Buffer.from(header), frame])); + // Even an empty data event resumes the reader's buffered input. + input.emit("data", Buffer.alloc(0)); + } + else { + input.write(header); + input.write(frame); + } + // Close synchronously, before the recovered frame is decoded or dispatched. + input.emit("close"); + assert.equal(await pending, "received"); + }); + } +} + +test("a close drains a fragmented response buffered after an invalid header", { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + const [message] = await request; + const frame = responseFrame(message.id, "received"); + input.write(Buffer.concat([Buffer.from("Content-Length: invalid\r\n\r\n"), frame.subarray(0, 10)])); + input.write(frame.subarray(10)); + input.emit("close"); + assert.equal(await pending, "received"); +}); + +test("a close rejects a response buffered behind an invalid header without another data event", { timeout: 10000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + const [message] = await request; + const rejected = assertRejectsPromptly(pending); + input.write(Buffer.concat([Buffer.from("Content-Length: invalid\r\n\r\n"), responseFrame(message.id, "received")])); + input.emit("close"); + await rejected; +}); + for (const invalidLength of ["invalid", "-1"]) { test(`a response followed by Content-Length ${invalidLength} does not prevent close`, { timeout: 10000 }, async t => { const { connection, input, received } = createWirePeer(t); From de7f146a360904969e5114d6444330d00f59dfc4 Mon Sep 17 00:00:00 2001 From: wsj-cmd Date: Wed, 4 Nov 2026 10:25:12 +0800 Subject: [PATCH 3/3] Match JSON-RPC header delimiter tracking to the reader --- packages/typescript/src/api/async/jsonrpc.ts | 19 ++++++- .../typescript/test/async/disconnect.test.ts | 51 +++++++++++++++++++ 2 files changed, 68 insertions(+), 2 deletions(-) diff --git a/packages/typescript/src/api/async/jsonrpc.ts b/packages/typescript/src/api/async/jsonrpc.ts index 75af6dc88cc98..ddb56cb0847f2 100644 --- a/packages/typescript/src/api/async/jsonrpc.ts +++ b/packages/typescript/src/api/async/jsonrpc.ts @@ -216,6 +216,7 @@ function responseKey(message: Message): string | undefined { /** Counts complete Content-Length frames, retaining only bytes deferred after malformed headers. */ class IncomingFrames { private header = ""; + private headerState = 0; private remaining: number | undefined; private deferred: Uint8Array | undefined; private valid = true; @@ -239,8 +240,22 @@ class IncomingFrames { } continue; } - this.header += String.fromCharCode(data[offset++] & 0x7f); - if (!this.header.endsWith("\r\n\r\n")) continue; + const value = data[offset++]; + this.header += String.fromCharCode(value & 0x7f); + // Match AbstractMessageBuffer.tryReadHeaders on raw bytes, before + // ASCII decoding, including its reset on overlapping CR sequences. + switch (value) { + case 13: + this.headerState = this.headerState === 0 ? 1 : this.headerState === 2 ? 3 : 0; + break; + case 10: + this.headerState = this.headerState === 1 ? 2 : this.headerState === 3 ? 4 : 0; + break; + default: + this.headerState = 0; + } + if (this.headerState !== 4) continue; + this.headerState = 0; const header = this.header; this.header = ""; // The reader consumes malformed headers, then retries the remaining diff --git a/packages/typescript/test/async/disconnect.test.ts b/packages/typescript/test/async/disconnect.test.ts index 4373fc0e0ad2c..91101f61b42b3 100644 --- a/packages/typescript/test/async/disconnect.test.ts +++ b/packages/typescript/test/async/disconnect.test.ts @@ -417,6 +417,57 @@ for ( }); } +for ( + const [description, data] of [ + ["overlapping CRLF", Buffer.from("Content-Length: 2\r\r\n\r\n{}")], + ["high-bit CRLF", Buffer.concat([Buffer.from("Content-Length: 1"), Buffer.from([0x8d, 0x8a, 0x8d, 0x8a, 0x7b])])], + ] as const +) { + for (const fragmented of [false, true]) { + test(`a close rejects a request after ${description} (${fragmented ? "fragmented" : "whole"})`, { timeout: 5000 }, async t => { + const { connection, input, received } = createWirePeer(t); + const request = once(received, "request"); + const pending = connection.sendRequest(new RequestType("probe"), {}); + await request; + const rejected = assertRejectsPromptly(pending); + const chunkSize = fragmented ? 1 : data.length; + for (let offset = 0; offset < data.length; offset += chunkSize) input.write(data.subarray(offset, offset + chunkSize)); + input.emit("close"); + await rejected; + }); + } + + test(`explicit client close completes after ${description} and disconnect`, { timeout: 5000 }, async t => { + const childProcess = process.getBuiltinModule("node:child_process"); + const child = new EventEmitter() as ChildProcess; + const input = new PassThrough(); + const output = new PassThrough(); + child.stdin = input; + child.stdout = output; + t.mock.method(childProcess, "spawn", () => { + process.nextTick(() => child.emit("spawn")); + return child; + }); + const client = createClient({ tsserverPath: process.execPath, useCaseSensitiveFileNames: true }); + t.after(() => { + input.destroy(); + output.destroy(); + }); + if (!("connect" in client)) throw new Error("Expected a process client"); + await client.connect(); + const reader = new StreamMessageReader(input); + const received = new Promise(resolve => reader.listen(() => resolve())); + t.after(() => reader.dispose()); + const pending = client.apiRequest("parseCommandLine", { commandLine: [] }); + await received; + const rejected = assertRejectsPromptly(pending); + output.write(data); + output.emit("close"); + await client.close(); + await rejected; + }); +} + test("writer close rejects pending requests without waiting for later input", { timeout: 10000 }, async t => { const { connection, input, output, received } = createWirePeer(t); const request = once(received, "request");