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/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 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/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 e977900f15e0783f75f1ef1b782396f2dc11e69e Mon Sep 17 00:00:00 2001 From: eleanorjboyd <26030610+eleanorjboyd@users.noreply.github.com> Date: Fri, 31 Jul 2026 11:24:54 -0700 Subject: [PATCH 3/3] Add POSIX result pipe cancellation integration test Exercise pytest cancellation through the real execution adapter and Python subprocess, preserving the completed result and requiring FIFO drainage to finish before the fallback timeout. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../testing/common/testingAdapter.test.ts | 71 ++++++++++++++++++- .../smallWorkspace/test_cancellation.py | 11 +++ 2 files changed, 81 insertions(+), 1 deletion(-) create mode 100644 src/testTestingRootWkspc/smallWorkspace/test_cancellation.py diff --git a/src/test/testing/common/testingAdapter.test.ts b/src/test/testing/common/testingAdapter.test.ts index d73efba2a8ec..dcc57c3d20ee 100644 --- a/src/test/testing/common/testingAdapter.test.ts +++ b/src/test/testing/common/testingAdapter.test.ts @@ -1,7 +1,7 @@ /* eslint-disable @typescript-eslint/no-explicit-any */ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. -import { TestController, TestRun, TestRunProfileKind, Uri } from 'vscode'; +import { CancellationTokenSource, TestController, TestRun, TestRunProfileKind, Uri } from 'vscode'; import * as typeMoq from 'typemoq'; import * as path from 'path'; import * as assert from 'assert'; @@ -694,6 +694,75 @@ suite('End to End Tests: test adapters', () => { }); }); + test('pytest cancellation closes the POSIX result pipe promptly', async () => { + if (process.platform === 'win32') { + return; + } + + workspaceUri = Uri.parse(rootPathSmallWorkspace); + configService.getSettings(workspaceUri).testing.pytestArgs = []; + + const cancellation = new CancellationTokenSource(); + const completedTestId = `${rootPathSmallWorkspace}/test_cancellation.py::test_result_before_cancellation`; + const slowTestId = `${rootPathSmallWorkspace}/test_cancellation.py::test_waits_for_cancellation`; + const previousRuntimeDir = process.env.XDG_RUNTIME_DIR; + const runtimeDir = fs.mkdtempSync(path.join(os.tmpdir(), 'vscode-python-cancellation-')); + process.env.XDG_RUNTIME_DIR = runtimeDir; + let completedResultReceived = false; + let cancellationStartedAt: number | undefined; + let watchdogTriggered = false; + const watchdog = setTimeout(() => { + watchdogTriggered = true; + cancellation.cancel(); + }, 60_000); + + resultResolver = new PythonResultResolver(testController, pytestProvider, workspaceUri); + resultResolver.resolveExecution = (payload, _runInstance) => { + const completedResult = Object.values(payload.result ?? {}).find( + (result) => result.test === completedTestId && result.outcome === 'success', + ); + if (completedResult && !cancellation.token.isCancellationRequested) { + completedResultReceived = true; + cancellationStartedAt = performance.now(); + cancellation.cancel(); + } + }; + + const testRun = typeMoq.Mock.ofType(); + testRun.setup((t) => t.token).returns(() => cancellation.token); + testRun.setup((t) => t.appendOutput(typeMoq.It.isAny())).returns(() => false); + + const executionAdapter = new PytestTestExecutionAdapter(configService, resultResolver, envVarsService); + try { + await executionAdapter.runTests( + workspaceUri, + [completedTestId, slowTestId], + TestRunProfileKind.Run, + testRun.object, + pythonExecFactory, + ); + } finally { + clearTimeout(watchdog); + cancellation.cancel(); + cancellation.dispose(); + if (previousRuntimeDir === undefined) { + delete process.env.XDG_RUNTIME_DIR; + } else { + process.env.XDG_RUNTIME_DIR = previousRuntimeDir; + } + fs.rmSync(runtimeDir, { force: true, recursive: true }); + } + + assert.ok(!watchdogTriggered, 'Expected pytest to report a result before the cancellation watchdog fired'); + assert.ok(completedResultReceived, 'Expected the completed test result before cancellation'); + assert.ok(cancellationStartedAt, 'Expected cancellation to start after receiving a completed test result'); + const cancellationDuration = performance.now() - cancellationStartedAt; + assert.ok( + cancellationDuration < 4_000, + `Expected cancellation to close the result pipe promptly, but it took ${cancellationDuration}ms`, + ); + }).timeout(75_000); + test('Unittest execution with coverage, small workspace', async () => { // result resolver and saved data for assertions resultResolver = new PythonResultResolver(testController, unittestProvider, workspaceUri); diff --git a/src/testTestingRootWkspc/smallWorkspace/test_cancellation.py b/src/testTestingRootWkspc/smallWorkspace/test_cancellation.py new file mode 100644 index 000000000000..ffa4ada51cfe --- /dev/null +++ b/src/testTestingRootWkspc/smallWorkspace/test_cancellation.py @@ -0,0 +1,11 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +import time + + +def test_result_before_cancellation(): + assert True + + +def test_waits_for_cancellation(): + time.sleep(120)