testing: Close POSIX result pipes on FIFO EOF

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<void> {
+        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<void>((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<rpc.MessageReader> {
     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);
 }

src/client/testing/testController/common/utils.ts

@@ -115,7 +115,7 @@ export async function startRunResultNamedPipe(
     dataReceivedCallback: (payload: ExecutionTestPayload) => void,
     deferredTillServerClose: Deferred<void>,
     cancellationToken?: CancellationToken,
-): Promise<string> {
+): 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 {

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();
         }
     }
 

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();
         }
     }
 

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<T>(promise: Promise<T>, description: string): Promise<T> {
+    let timeout: NodeJS.Timeout | undefined;
+    try {
+        return await Promise.race([
+            promise,
+            new Promise<T>((_, 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<void>();
+        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<void>();
+        const listener = reader.listen(() => undefined);
+        const closeListener = reader.onClose(() => closed.resolve());
+        const stream = fs.createWriteStream(pipeName);
+
+        try {
+            await new Promise<void>((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<void>();
+        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 });
+        }
+    });
+});

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'));
     });
@@ -186,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<TestRun>();

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);
         });
     });
 });