Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 28 additions & 2 deletions src/client/testing/testController/common/utils.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,11 @@ export function createTestingDeferred(): Deferred<void> {
// platform-specific named-pipe quirks) so the adapter's `finally` block can never hang.
export const RESULT_PIPE_DRAIN_TIMEOUT_MS = 5_000;

// Grace period (ms) to allow buffered results to drain after cancellation
// or subprocess termination before closing the result pipe when `reader.onClose`
// does not fire (such as non-blocking FIFO sockets on macOS).
export const RESULT_PIPE_DRAIN_GRACE_PERIOD_MS = 100;

/**
* Awaits `deferred.promise` but resolves after at most `timeoutMs` so callers
* cannot hang indefinitely if the underlying event source never fires.
Expand Down Expand Up @@ -119,18 +124,39 @@ export async function startRunResultNamedPipe(
traceVerbose('Starting Test Result named pipe');
const pipeName: string = generateRandomPipeName('python-test-results');

// `cancellationToken` only cancels pipe creation; disposal is driven by
// `reader.onClose` so buffered results are not dropped on cancel.
// `cancellationToken` cancels pipe creation immediately. On cancellation or subprocess
// completion, a brief drain window is started so buffered messages are delivered before
// disposal when `reader.onClose` never fires (e.g. non-blocking FIFOs on macOS).
const reader = await createReaderPipe(pipeName, cancellationToken);
traceVerbose(`Test Results named pipe ${pipeName} connected`);
let drainTimer: NodeJS.Timeout | undefined;
let disposables: Disposable[] = [];
const disposable = new Disposable(() => {
if (drainTimer) {
clearTimeout(drainTimer);
drainTimer = undefined;
}
traceVerbose(`Test Results named pipe ${pipeName} disposed`);
disposables.forEach((d) => d.dispose());
disposables = [];
deferredTillServerClose.resolve();
});

if (cancellationToken) {
const scheduleDrain = () => {
if (!drainTimer) {
drainTimer = setTimeout(() => {
disposable.dispose();
}, RESULT_PIPE_DRAIN_GRACE_PERIOD_MS);
}
};
if (cancellationToken.isCancellationRequested) {
scheduleDrain();
} else {
disposables.push(cancellationToken.onCancellationRequested(scheduleDrain));
}
}

disposables.push(
reader,
reader.listen((data: Message) => {
Expand Down
22 changes: 22 additions & 0 deletions src/test/testing/testController/utils.unit.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -935,6 +935,28 @@ suite('startRunResultNamedPipe drain-on-cancel tests', () => {

assert.strictEqual(reader.disposed, true, 'reader disposed when onClose fires');
});

test('cancellation with reader that never closes resolves deferredTillServerClose after drain grace period', async () => {
const reader = new FakeMessageReader();
sandbox.stub(namedPipes, 'createReaderPipe').resolves(reader);

const received: ExecutionTestPayload[] = [];
const deferredTillServerClose: Deferred<void> = createDeferred<void>();
const cancelSource = new CancellationTokenSource();

await startRunResultNamedPipe((payload) => received.push(payload), deferredTillServerClose, cancelSource.token);

cancelSource.cancel();
// Before grace period expires, not yet resolved
await new Promise((r) => setImmediate(r));
assert.strictEqual((deferredTillServerClose as Deferred<void>).completed, false);

// After grace period expires, resolved and reader disposed
await deferredTillServerClose.promise;
assert.strictEqual(reader.disposed, true);

cancelSource.dispose();
});
});

suite('awaitDeferredWithTimeout', () => {
Expand Down