From d741dd3ea451f0caf098bb2ee9ca6ba204012205 Mon Sep 17 00:00:00 2001 From: eleanorjboyd <26030610+eleanorjboyd@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:40:53 -0700 Subject: [PATCH 1/3] Fix POSIX result pipe completion Drain nonblocking FIFO readers to EOF and explicitly dispose result pipes after the fallback timeout. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/client/common/pipes/namedPipes.ts | 118 +++++++++++++++-- .../testing/testController/common/utils.ts | 7 +- .../pytest/pytestExecutionAdapter.ts | 5 +- .../unittest/testExecutionAdapter.ts | 5 +- src/test/common/pipes/namedPipes.unit.test.ts | 119 ++++++++++++++++++ .../pytestExecutionAdapter.unit.test.ts | 4 +- .../testCancellationRunAdapters.unit.test.ts | 6 +- .../testExecutionAdapter.unit.test.ts | 6 +- 8 files changed, 251 insertions(+), 19 deletions(-) create mode 100644 src/test/common/pipes/namedPipes.unit.test.ts diff --git a/src/client/common/pipes/namedPipes.ts b/src/client/common/pipes/namedPipes.ts index 9bffe78f2b9f..ccad22da1249 100644 --- a/src/client/common/pipes/namedPipes.ts +++ b/src/client/common/pipes/namedPipes.ts @@ -7,6 +7,7 @@ import * as fs from 'fs-extra'; import * as net from 'net'; import * as os from 'os'; import * as path from 'path'; +import { PassThrough } from 'stream'; import * as rpc from 'vscode-jsonrpc/node'; import { CancellationError, CancellationToken, Disposable } from 'vscode'; import { traceVerbose } from '../../logging'; @@ -15,6 +16,9 @@ import { createDeferred } from '../utils/async'; import { noop } from '../utils/misc'; const { XDG_RUNTIME_DIR } = process.env; +const FIFO_READ_RETRY_DELAY_MS = 10; +const FIFO_READ_BUFFER_SIZE = 64 * 1024; + export function generateRandomPipeName(prefix: string): string { // length of 10 picked because of the name length restriction for sockets const randomSuffix = crypto.randomBytes(10).toString('hex'); @@ -149,6 +153,111 @@ class CombinedReader implements rpc.MessageReader { } } +// A net.Socket does not surface EOF for a nonblocking FIFO descriptor, so read +// the descriptor directly and close after observing its writer disconnect. +class FifoMessageReader extends rpc.AbstractMessageReader { + private readonly stream = new PassThrough(); + + private readonly messageReader = new rpc.StreamMessageReader(this.stream, 'utf-8'); + + private readonly buffer = Buffer.allocUnsafe(FIFO_READ_BUFFER_SIZE); + + private readonly disposables: rpc.Disposable[] = []; + + private hasReadData = false; + + private hasObservedWriter = false; + + private disposed = false; + + private closePending = false; + + constructor( + private readonly pipeName: string, + private readonly fd: number, + private readonly token?: CancellationToken, + ) { + super(); + this.disposables.push( + this.messageReader.onError((error) => this.fireError(error)), + this.messageReader.onPartialMessage((info) => this.firePartialMessage(info)), + ); + } + + listen(callback: rpc.DataCallback): rpc.Disposable { + const listener = this.messageReader.listen(callback); + void this.read(); + return listener; + } + + private async read(): Promise { + while (!this.disposed && !this.closePending) { + try { + const { bytesRead } = await fs.read(this.fd, this.buffer, 0, this.buffer.length, null); + if (bytesRead > 0) { + this.hasReadData = true; + this.stream.write(Buffer.from(this.buffer.subarray(0, bytesRead))); + continue; + } + + if (this.hasReadData || this.hasObservedWriter || this.token?.isCancellationRequested) { + this.complete(); + return; + } + } catch (error) { + if (this.isRetryableReadError(error)) { + this.hasObservedWriter = true; + } else { + if (!this.disposed) { + this.fireError(error); + this.complete(); + } + return; + } + } + + await new Promise((resolve) => { + setTimeout(resolve, FIFO_READ_RETRY_DELAY_MS); + }); + } + } + + private isRetryableReadError(error: unknown): boolean { + if (!(error instanceof Error) || !('code' in error)) { + return false; + } + const { code } = error as NodeJS.ErrnoException; + return code === 'EAGAIN' || code === 'EWOULDBLOCK'; + } + + private complete(): void { + if (this.disposed || this.closePending) { + return; + } + this.closePending = true; + this.stream.end(); + setImmediate(() => { + if (!this.disposed) { + this.fireClose(); + } + }); + } + + dispose(): void { + if (this.disposed) { + return; + } + this.disposed = true; + this.stream.destroy(); + this.messageReader.dispose(); + this.disposables.forEach((disposable) => disposable.dispose()); + this.disposables.length = 0; + fs.close(this.fd).catch(noop); + fs.unlink(this.pipeName).catch(noop); + super.dispose(); + } +} + export async function createReaderPipe(pipeName: string, token?: CancellationToken): Promise { if (isWindows()) { // windows implementation of FIFO using named pipes @@ -189,12 +298,5 @@ export async function createReaderPipe(pipeName: string, token?: CancellationTok // Intentionally ignored } const fd = await fs.open(pipeName, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); - const socket = new net.Socket({ fd }); - const reader = new rpc.SocketMessageReader(socket, 'utf-8'); - socket.on('close', () => { - fs.close(fd).catch(noop); - reader.dispose(); - }); - - return reader; + return new FifoMessageReader(pipeName, fd, token); } diff --git a/src/client/testing/testController/common/utils.ts b/src/client/testing/testController/common/utils.ts index 583d57648548..ecba099968ab 100644 --- a/src/client/testing/testController/common/utils.ts +++ b/src/client/testing/testController/common/utils.ts @@ -115,7 +115,7 @@ export async function startRunResultNamedPipe( dataReceivedCallback: (payload: ExecutionTestPayload) => void, deferredTillServerClose: Deferred, cancellationToken?: CancellationToken, -): Promise { +): Promise<{ name: string; dispose: () => void }> { traceVerbose('Starting Test Result named pipe'); const pipeName: string = generateRandomPipeName('python-test-results'); @@ -147,7 +147,10 @@ export async function startRunResultNamedPipe( }), ); - return pipeName; + return { + name: pipeName, + dispose: () => disposable.dispose(), + }; } interface DiscoveryResultMessage extends Message { diff --git a/src/client/testing/testController/pytest/pytestExecutionAdapter.ts b/src/client/testing/testController/pytest/pytestExecutionAdapter.ts index a982037d90e5..ea60297b1ffe 100644 --- a/src/client/testing/testController/pytest/pytestExecutionAdapter.ts +++ b/src/client/testing/testController/pytest/pytestExecutionAdapter.ts @@ -53,7 +53,7 @@ export class PytestTestExecutionAdapter implements ITestExecutionAdapter { const cSource = new CancellationTokenSource(); runInstance.token.onCancellationRequested(() => cSource.cancel()); - const name = await utils.startRunResultNamedPipe( + const resultPipe = await utils.startRunResultNamedPipe( dataReceivedCallback, // callback to handle data received deferredTillServerClose, // deferred to resolve when server closes cSource.token, // token to cancel @@ -66,7 +66,7 @@ export class PytestTestExecutionAdapter implements ITestExecutionAdapter { await this.runTestsNew( uri, testIds, - name, + resultPipe.name, cSource, runInstance, profileKind, @@ -77,6 +77,7 @@ export class PytestTestExecutionAdapter implements ITestExecutionAdapter { ); } finally { await utils.awaitDeferredWithTimeout(deferredTillServerClose, utils.RESULT_PIPE_DRAIN_TIMEOUT_MS); + resultPipe.dispose(); } } diff --git a/src/client/testing/testController/unittest/testExecutionAdapter.ts b/src/client/testing/testController/unittest/testExecutionAdapter.ts index 755da2e3c65a..2061a6e948a6 100644 --- a/src/client/testing/testController/unittest/testExecutionAdapter.ts +++ b/src/client/testing/testController/unittest/testExecutionAdapter.ts @@ -64,7 +64,7 @@ export class UnittestTestExecutionAdapter implements ITestExecutionAdapter { }; const cSource = new CancellationTokenSource(); runInstance.token.onCancellationRequested(() => cSource.cancel()); - const name = await utils.startRunResultNamedPipe( + const resultPipe = await utils.startRunResultNamedPipe( dataReceivedCallback, // callback to handle data received deferredTillServerClose, // deferred to resolve when server closes cSource.token, // token to cancel @@ -79,7 +79,7 @@ export class UnittestTestExecutionAdapter implements ITestExecutionAdapter { await this.runTestsNew( uri, testIds, - name, + resultPipe.name, cSource, runInstance, profileKind, @@ -91,6 +91,7 @@ export class UnittestTestExecutionAdapter implements ITestExecutionAdapter { traceError(`Error in running unittest tests: ${error}`); } finally { await utils.awaitDeferredWithTimeout(deferredTillServerClose, utils.RESULT_PIPE_DRAIN_TIMEOUT_MS); + resultPipe.dispose(); } } diff --git a/src/test/common/pipes/namedPipes.unit.test.ts b/src/test/common/pipes/namedPipes.unit.test.ts new file mode 100644 index 000000000000..981107bdfce7 --- /dev/null +++ b/src/test/common/pipes/namedPipes.unit.test.ts @@ -0,0 +1,119 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as assert from 'assert'; +import * as fs from 'fs'; +import { CancellationTokenSource } from 'vscode'; +import * as rpc from 'vscode-jsonrpc/node'; +import { createDeferred } from '../../../client/common/utils/async'; +import { createReaderPipe, generateRandomPipeName } from '../../../client/common/pipes/namedPipes'; + +const TEST_TIMEOUT_MS = 2_000; + +async function waitFor(promise: Promise, description: string): Promise { + let timeout: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error(`Timed out waiting for ${description}`)), TEST_TIMEOUT_MS); + }), + ]); + } finally { + if (timeout) { + clearTimeout(timeout); + } + } +} + +suite('POSIX named pipe reader', () => { + test('delivers buffered messages and closes when the FIFO writer closes', async function () { + if (process.platform === 'win32') { + this.skip(); + } + + const pipeName = generateRandomPipeName('python-test-fifo-reader'); + const reader = await createReaderPipe(pipeName); + const received: rpc.Message[] = []; + const closed = createDeferred(); + const listener = reader.listen((message) => received.push(message)); + const closeListener = reader.onClose(() => closed.resolve()); + const stream = fs.createWriteStream(pipeName); + const writer = new rpc.StreamMessageWriter(stream, 'utf-8'); + const expected: rpc.NotificationMessage[] = Array.from({ length: 20 }, (_, value) => ({ + jsonrpc: '2.0', + method: 'test/message', + params: { value, data: 'x'.repeat(4_096) }, + })); + + try { + for (const message of expected) { + await writer.write(message); + } + writer.end(); + + await waitFor(closed.promise, 'the FIFO reader to close'); + + assert.deepStrictEqual(received, expected); + } finally { + listener.dispose(); + closeListener.dispose(); + reader.dispose(); + writer.dispose(); + await fs.promises.rm(pipeName, { force: true }); + } + }); + + test('closes when a connected FIFO writer sends no messages', async function () { + if (process.platform === 'win32') { + this.skip(); + } + + const pipeName = generateRandomPipeName('python-test-fifo-empty'); + const reader = await createReaderPipe(pipeName); + const closed = createDeferred(); + const listener = reader.listen(() => undefined); + const closeListener = reader.onClose(() => closed.resolve()); + const stream = fs.createWriteStream(pipeName); + + try { + await new Promise((resolve, reject) => { + stream.once('open', () => setTimeout(resolve, 50)); + stream.once('error', reject); + }); + stream.end(); + + await waitFor(closed.promise, 'the empty FIFO reader to close'); + } finally { + listener.dispose(); + closeListener.dispose(); + reader.dispose(); + stream.destroy(); + await fs.promises.rm(pipeName, { force: true }); + } + }); + + test('closes on cancellation when no FIFO writer connected', async function () { + if (process.platform === 'win32') { + this.skip(); + } + + const pipeName = generateRandomPipeName('python-test-fifo-cancel'); + const cancellation = new CancellationTokenSource(); + const reader = await createReaderPipe(pipeName, cancellation.token); + const closed = createDeferred(); + const listener = reader.listen(() => undefined); + const closeListener = reader.onClose(() => closed.resolve()); + + try { + cancellation.cancel(); + await waitFor(closed.promise, 'the cancelled FIFO reader to close'); + } finally { + listener.dispose(); + closeListener.dispose(); + reader.dispose(); + cancellation.dispose(); + await fs.promises.rm(pipeName, { force: true }); + } + }); +}); diff --git a/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts b/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts index 9b83f1bbd088..c0083690e0a5 100644 --- a/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts +++ b/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts @@ -89,7 +89,9 @@ suite('pytest test execution adapter', () => { myTestPath = path.join('/', 'my', 'test', 'path', '/'); utilsStartRunResultNamedPipeStub = sinon.stub(util, 'startRunResultNamedPipe'); - utilsStartRunResultNamedPipeStub.callsFake(() => Promise.resolve('runResultPipe-mockName')); + utilsStartRunResultNamedPipeStub.callsFake(() => + Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() }), + ); execService.setup((x) => x.getExecutablePath()).returns(() => Promise.resolve('/mock/path/to/python')); }); diff --git a/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts b/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts index cdf0d00c5dc4..899308f81b13 100644 --- a/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts +++ b/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts @@ -112,7 +112,7 @@ suite('Execution Flow Run Adapters', () => { deferredTillServerCloseTester?.resolve(); }); - return Promise.resolve('named-pipes-socket-name'); + return Promise.resolve({ name: 'named-pipes-socket-name', dispose: serverDisposeStub }); }); serverDisposeStub.callsFake(() => { console.log('server disposed'); @@ -138,6 +138,7 @@ suite('Execution Flow Run Adapters', () => { ); // wait for server to start to keep test from failing await deferredStartTestIdsNamedPipe.promise; + sinon.assert.calledOnce(serverDisposeStub); }); test(`Adapter ${adapter}: token called mid-debug resolves correctly`, async () => { // mock test run and cancelation token @@ -180,7 +181,7 @@ suite('Execution Flow Run Adapters', () => { token?.onCancellationRequested(() => { deferredTillServerCloseTester?.resolve(); }); - return Promise.resolve('named-pipes-socket-name'); + return Promise.resolve({ name: 'named-pipes-socket-name', dispose: serverDisposeStub }); }); serverDisposeStub.callsFake(() => { console.log('server disposed'); @@ -219,6 +220,7 @@ suite('Execution Flow Run Adapters', () => { ); // wait for server to start to keep test from failing await deferredStartTestIdsNamedPipe.promise; + sinon.assert.calledOnce(serverDisposeStub); }); }); }); diff --git a/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts b/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts index 8a86e9228567..a8d40bad477f 100644 --- a/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts +++ b/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts @@ -89,7 +89,9 @@ suite('Unittest test execution adapter', () => { myTestPath = path.join('/', 'my', 'test', 'path', '/'); utilsStartRunResultNamedPipeStub = sinon.stub(util, 'startRunResultNamedPipe'); - utilsStartRunResultNamedPipeStub.callsFake(() => Promise.resolve('runResultPipe-mockName')); + utilsStartRunResultNamedPipeStub.callsFake(() => + Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() }), + ); execService.setup((x) => x.getExecutablePath()).returns(() => Promise.resolve('/mock/path/to/python')); }); @@ -510,7 +512,7 @@ suite('Unittest test execution adapter', () => { let serverCloseDeferred: Deferred | undefined; utilsStartRunResultNamedPipeStub.callsFake((_callback: unknown, deferred: Deferred, _token: unknown) => { serverCloseDeferred = deferred; - return Promise.resolve('runResultPipe-mockName'); + return Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() }); }); const projectPath = path.join('/', 'workspace', 'myproject'); From 5f5761f465d747cde6ecdc5c31abe6c6ff04e73e Mon Sep 17 00:00:00 2001 From: eleanorjboyd <26030610+eleanorjboyd@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:45:47 -0700 Subject: [PATCH 2/3] Update pytest result pipe test mock Use the disposable result pipe handle returned by the updated test utility. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../testController/pytest/pytestExecutionAdapter.unit.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts b/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts index c0083690e0a5..f7755be1b5f5 100644 --- a/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts +++ b/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts @@ -188,7 +188,7 @@ suite('pytest test execution adapter', () => { utilsWriteTestIdsFileStub.resolves('testIdPipe-mockName'); utilsStartRunResultNamedPipeStub.callsFake((_callback, deferredTillServerClose, token) => { token?.onCancellationRequested(() => deferredTillServerClose.resolve()); - return Promise.resolve('runResultPipe-mockName'); + return Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() }); }); const cancellationToken = new CancellationTokenSource(); const testRun = typeMoq.Mock.ofType(); From eec9b727a98adf46e418e1c23ffadb7754c66472 Mon Sep 17 00:00:00 2001 From: eleanorjboyd <26030610+eleanorjboyd@users.noreply.github.com> Date: Tue, 29 Sep 2026 13:33:28 -0700 Subject: [PATCH 3/3] Harden POSIX result pipe cleanup Honor parser backpressure and signal pipe completion when test subprocess work ends so empty or failed writers cannot leave polling readers behind. Add focused coverage for backpressure and empty-writer cleanup.\n\nCo-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/client/common/pipes/namedPipes.ts | 18 ++++- .../pytest/pytestDiscoveryAdapter.ts | 3 +- .../pytest/pytestExecutionAdapter.ts | 2 + .../unittest/testDiscoveryAdapter.ts | 3 +- .../unittest/testExecutionAdapter.ts | 2 + src/test/common/pipes/namedPipes.unit.test.ts | 80 +++++++++++++++++-- 6 files changed, 98 insertions(+), 10 deletions(-) diff --git a/src/client/common/pipes/namedPipes.ts b/src/client/common/pipes/namedPipes.ts index ccad22da1249..de8fb07bfd99 100644 --- a/src/client/common/pipes/namedPipes.ts +++ b/src/client/common/pipes/namedPipes.ts @@ -154,7 +154,7 @@ class CombinedReader implements rpc.MessageReader { } // A net.Socket does not surface EOF for a nonblocking FIFO descriptor, so read -// the descriptor directly and close after observing its writer disconnect. +// the descriptor directly and close after EOF or owner-signaled completion. class FifoMessageReader extends rpc.AbstractMessageReader { private readonly stream = new PassThrough(); @@ -196,7 +196,9 @@ class FifoMessageReader extends rpc.AbstractMessageReader { const { bytesRead } = await fs.read(this.fd, this.buffer, 0, this.buffer.length, null); if (bytesRead > 0) { this.hasReadData = true; - this.stream.write(Buffer.from(this.buffer.subarray(0, bytesRead))); + if (!this.stream.write(Buffer.from(this.buffer.subarray(0, bytesRead)))) { + await this.waitForDrain(); + } continue; } @@ -222,6 +224,18 @@ class FifoMessageReader extends rpc.AbstractMessageReader { } } + private waitForDrain(): Promise { + return new Promise((resolve) => { + const finish = () => { + this.stream.off('drain', finish); + this.stream.off('close', finish); + resolve(); + }; + this.stream.once('drain', finish); + this.stream.once('close', finish); + }); + } + private isRetryableReadError(error: unknown): boolean { if (!(error instanceof Error) || !('code' in error)) { return false; diff --git a/src/client/testing/testController/pytest/pytestDiscoveryAdapter.ts b/src/client/testing/testController/pytest/pytestDiscoveryAdapter.ts index 16e27635e66c..af002ad03b18 100644 --- a/src/client/testing/testController/pytest/pytestDiscoveryAdapter.ts +++ b/src/client/testing/testController/pytest/pytestDiscoveryAdapter.ts @@ -218,7 +218,8 @@ export class PytestTestDiscoveryAdapter implements ITestDiscoveryAdapter { } finally { // Dispose all cancellation handlers and event subscriptions disposables.forEach((d) => d.dispose()); - // Dispose the discovery pipe cancellation token + // Stop the discovery pipe even when the subprocess never opened it. + discoveryPipeCancellation.cancel(); discoveryPipeCancellation.dispose(); } } diff --git a/src/client/testing/testController/pytest/pytestExecutionAdapter.ts b/src/client/testing/testController/pytest/pytestExecutionAdapter.ts index ea60297b1ffe..2443c45a6527 100644 --- a/src/client/testing/testController/pytest/pytestExecutionAdapter.ts +++ b/src/client/testing/testController/pytest/pytestExecutionAdapter.ts @@ -76,8 +76,10 @@ export class PytestTestExecutionAdapter implements ITestExecutionAdapter { project, ); } finally { + cSource.cancel(); await utils.awaitDeferredWithTimeout(deferredTillServerClose, utils.RESULT_PIPE_DRAIN_TIMEOUT_MS); resultPipe.dispose(); + cSource.dispose(); } } diff --git a/src/client/testing/testController/unittest/testDiscoveryAdapter.ts b/src/client/testing/testController/unittest/testDiscoveryAdapter.ts index 558e01f3514d..91583fa49444 100644 --- a/src/client/testing/testController/unittest/testDiscoveryAdapter.ts +++ b/src/client/testing/testController/unittest/testDiscoveryAdapter.ts @@ -204,7 +204,8 @@ export class UnittestTestDiscoveryAdapter implements ITestDiscoveryAdapter { traceVerbose(`Cleaning up unittest discovery resources for workspace ${uri.fsPath}`); // Dispose all cancellation handlers and event subscriptions disposables.forEach((d) => d.dispose()); - // Dispose the discovery pipe cancellation token + // Stop the discovery pipe even when the subprocess never opened it. + discoveryPipeCancellation.cancel(); discoveryPipeCancellation.dispose(); } } diff --git a/src/client/testing/testController/unittest/testExecutionAdapter.ts b/src/client/testing/testController/unittest/testExecutionAdapter.ts index 2061a6e948a6..b1649de746f1 100644 --- a/src/client/testing/testController/unittest/testExecutionAdapter.ts +++ b/src/client/testing/testController/unittest/testExecutionAdapter.ts @@ -90,8 +90,10 @@ export class UnittestTestExecutionAdapter implements ITestExecutionAdapter { } catch (error) { traceError(`Error in running unittest tests: ${error}`); } finally { + cSource.cancel(); await utils.awaitDeferredWithTimeout(deferredTillServerClose, utils.RESULT_PIPE_DRAIN_TIMEOUT_MS); resultPipe.dispose(); + cSource.dispose(); } } diff --git a/src/test/common/pipes/namedPipes.unit.test.ts b/src/test/common/pipes/namedPipes.unit.test.ts index 981107bdfce7..e72b069566cf 100644 --- a/src/test/common/pipes/namedPipes.unit.test.ts +++ b/src/test/common/pipes/namedPipes.unit.test.ts @@ -1,8 +1,11 @@ +/* eslint-disable @typescript-eslint/no-explicit-any */ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. import * as assert from 'assert'; import * as fs from 'fs'; +import { PassThrough } from 'stream'; +import * as sinon from 'sinon'; import { CancellationTokenSource } from 'vscode'; import * as rpc from 'vscode-jsonrpc/node'; import { createDeferred } from '../../../client/common/utils/async'; @@ -64,30 +67,95 @@ suite('POSIX named pipe reader', () => { } }); - test('closes when a connected FIFO writer sends no messages', async function () { + test('pauses FIFO reads while the parser stream applies backpressure', async function () { if (process.platform === 'win32') { this.skip(); } - const pipeName = generateRandomPipeName('python-test-fifo-empty'); + const originalWrite = PassThrough.prototype.write; + let drainListener: (() => void) | undefined; + const waitingForDrain = createDeferred(); + const writeStub = sinon.stub(PassThrough.prototype, 'write'); + writeStub.onFirstCall().callsFake(function (this: PassThrough, ...args: any[]) { + originalWrite.apply(this, args as any); + return false; + }); + writeStub.callThrough(); + + const originalOnce = PassThrough.prototype.once; + const onceStub = sinon + .stub(PassThrough.prototype, 'once') + .callsFake(function (this: PassThrough, event: string | symbol, listener: (...args: any[]) => void) { + if (event === 'drain' && !drainListener) { + drainListener = listener; + waitingForDrain.resolve(); + return this; + } + return originalOnce.call(this, event, listener); + }); + + const pipeName = generateRandomPipeName('python-test-fifo-backpressure'); const reader = await createReaderPipe(pipeName); + const received = createDeferred(); + const closed = createDeferred(); + const listener = reader.listen((message) => received.resolve(message)); + const closeListener = reader.onClose(() => closed.resolve()); + const stream = fs.createWriteStream(pipeName); + const writer = new rpc.StreamMessageWriter(stream, 'utf-8'); + const expected: rpc.NotificationMessage = { + jsonrpc: '2.0', + method: 'test/message', + params: { data: 'x'.repeat(256 * 1_024) }, + }; + + try { + const writePromise = writer.write(expected); + await waitFor(waitingForDrain.promise, 'the FIFO reader to observe backpressure'); + + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.strictEqual(received.completed, false); + + drainListener?.(); + await writePromise; + writer.end(); + + await waitFor(closed.promise, 'the backpressured FIFO reader to close'); + assert.deepStrictEqual(await received.promise, expected); + } finally { + listener.dispose(); + closeListener.dispose(); + reader.dispose(); + writer.dispose(); + writeStub.restore(); + onceStub.restore(); + await fs.promises.rm(pipeName, { force: true }); + } + }); + + test('closes on completion when a connected FIFO writer sends no messages', async function () { + if (process.platform === 'win32') { + this.skip(); + } + + const pipeName = generateRandomPipeName('python-test-fifo-empty'); + const cancellation = new CancellationTokenSource(); + const reader = await createReaderPipe(pipeName, cancellation.token); const closed = createDeferred(); const listener = reader.listen(() => undefined); const closeListener = reader.onClose(() => closed.resolve()); const stream = fs.createWriteStream(pipeName); try { - await new Promise((resolve, reject) => { - stream.once('open', () => setTimeout(resolve, 50)); - stream.once('error', reject); - }); + await new Promise((resolve, reject) => stream.once('open', resolve).once('error', reject)); stream.end(); + cancellation.cancel(); await waitFor(closed.promise, 'the empty FIFO reader to close'); } finally { listener.dispose(); closeListener.dispose(); reader.dispose(); + cancellation.dispose(); stream.destroy(); await fs.promises.rm(pipeName, { force: true }); }