From 212a6bf8fc318e544f3b26efe6639d3c63335d0c 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/2] 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 0edcb77e1e29f8bd8736cb55b67886dee1a8c9f7 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/2] 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();