From 1216b5af96065b300df051227d1641f80b4106a6 Mon Sep 17 00:00:00 2001 From: Vladimir Date: Fri, 7 Aug 2026 14:34:48 +0200 Subject: [PATCH] fix: don't lose worker output on teardown, deflake timing-sensitive tests (#10842) --- .../src/node/pools/workers/forksWorker.ts | 13 ++++--- .../src/node/pools/workers/threadsWorker.ts | 24 +++++++++---- .../vitest/src/node/pools/workers/utils.ts | 14 ++++++++ packages/vitest/src/runtime/workers/init.ts | 29 +++++++++++++++ .../fixtures/server-url/vitest.config.ts | 5 ++- test/browser/specs/server-url.test.ts | 4 +-- test/e2e/test/concurrent.test.ts | 36 ++++++++++++++----- .../test/reporters/import-durations.test.ts | 14 +++++--- test/e2e/test/server-url.test.ts | 15 +++++--- 9 files changed, 124 insertions(+), 30 deletions(-) create mode 100644 packages/vitest/src/node/pools/workers/utils.ts diff --git a/packages/vitest/src/node/pools/workers/forksWorker.ts b/packages/vitest/src/node/pools/workers/forksWorker.ts index a95c1b041..a7cda6a83 100644 --- a/packages/vitest/src/node/pools/workers/forksWorker.ts +++ b/packages/vitest/src/node/pools/workers/forksWorker.ts @@ -3,6 +3,7 @@ import type { Writable } from 'node:stream' import type { PoolOptions, PoolWorker, WorkerRequest } from '../types' import { fork } from 'node:child_process' import { resolve } from 'node:path' +import { streamFlushed } from './utils' const SIGKILL_TIMEOUT = 500 /** jest does 500ms by default, let's follow it */ @@ -49,14 +50,16 @@ export class ForksPoolWorker implements PoolWorker { serialization: 'advanced', }) + // `end: false`: the logger streams are shared by every worker, so one + // ending worker stream must not end them for everyone else if (this._fork.stdout) { this.stdout.setMaxListeners(1 + this.stdout.getMaxListeners()) - this._fork.stdout.pipe(this.stdout) + this._fork.stdout.pipe(this.stdout, { end: false }) } if (this._fork.stderr) { this.stderr.setMaxListeners(1 + this.stderr.getMaxListeners()) - this._fork.stderr.pipe(this.stderr) + this._fork.stderr.pipe(this.stderr, { end: false }) } } @@ -87,12 +90,14 @@ export class ForksPoolWorker implements PoolWorker { clearTimeout(sigkillTimeout) if (fork.stdout) { - fork.stdout?.unpipe(this.stdout) + await streamFlushed(fork.stdout) + fork.stdout.unpipe(this.stdout) this.stdout.setMaxListeners(this.stdout.getMaxListeners() - 1) } if (fork.stderr) { - fork.stderr?.unpipe(this.stderr) + await streamFlushed(fork.stderr) + fork.stderr.unpipe(this.stderr) this.stderr.setMaxListeners(this.stderr.getMaxListeners() - 1) } diff --git a/packages/vitest/src/node/pools/workers/threadsWorker.ts b/packages/vitest/src/node/pools/workers/threadsWorker.ts index 05609a0b2..3fb111a48 100644 --- a/packages/vitest/src/node/pools/workers/threadsWorker.ts +++ b/packages/vitest/src/node/pools/workers/threadsWorker.ts @@ -2,6 +2,7 @@ import type { Writable } from 'node:stream' import type { PoolOptions, PoolWorker, WorkerRequest } from '../types' import { resolve } from 'node:path' import { Worker } from 'node:worker_threads' +import { streamFlushed } from './utils' /** @experimental */ export class ThreadsPoolWorker implements PoolWorker { @@ -46,20 +47,31 @@ export class ThreadsPoolWorker implements PoolWorker { stderr: true, }) + // `end: false`: the logger streams are shared by every worker, so one + // ending worker stream must not end them for everyone else this.stdout.setMaxListeners(1 + this.stdout.getMaxListeners()) - this._thread.stdout.pipe(this.stdout) + this._thread.stdout.pipe(this.stdout, { end: false }) this.stderr.setMaxListeners(1 + this.stderr.getMaxListeners()) - this._thread.stderr.pipe(this.stderr) + this._thread.stderr.pipe(this.stderr, { end: false }) } async stop(): Promise { - await this.thread.terminate() - - this._thread?.stdout?.unpipe(this.stdout) + const thread = this.thread + // `terminate()` makes node drain the stdio still queued on the worker's + // message port into these readables; keep the pipes attached until the + // streams end so late output still reaches the logger streams + const flushed = Promise.all([ + streamFlushed(thread.stdout), + streamFlushed(thread.stderr), + ]) + await thread.terminate() + await flushed + + thread.stdout.unpipe(this.stdout) this.stdout.setMaxListeners(this.stdout.getMaxListeners() - 1) - this._thread?.stderr?.unpipe(this.stderr) + thread.stderr.unpipe(this.stderr) this.stderr.setMaxListeners(this.stderr.getMaxListeners() - 1) this._thread = undefined diff --git a/packages/vitest/src/node/pools/workers/utils.ts b/packages/vitest/src/node/pools/workers/utils.ts new file mode 100644 index 000000000..9e3c786d3 --- /dev/null +++ b/packages/vitest/src/node/pools/workers/utils.ts @@ -0,0 +1,14 @@ +import type { Readable } from 'node:stream' + +// After a worker dies, its remaining stdio is drained into the parent-side +// readables asynchronously. Waiting for `end`/`close` before unpiping ensures +// the tail of the output still reaches the shared logger streams. +export function streamFlushed(stream: Readable): Promise { + if (stream.readableEnded || stream.destroyed) { + return Promise.resolve() + } + return new Promise((resolve) => { + stream.once('end', resolve) + stream.once('close', resolve) + }) +} diff --git a/packages/vitest/src/runtime/workers/init.ts b/packages/vitest/src/runtime/workers/init.ts index 7316273d1..7cbdb8432 100644 --- a/packages/vitest/src/runtime/workers/init.ts +++ b/packages/vitest/src/runtime/workers/init.ts @@ -49,6 +49,27 @@ const __vitest_worker_response__ = true const memoryUsage = process.memoryUsage.bind(process) let reportMemory = false +// In worker threads stdio is proxied to the parent over a MessagePort with a +// backpressure protocol: a chunk stays buffered inside the worker until the +// parent acks the previous one. The pool starts `runner.stop()` as soon as it +// receives `testfileFinished`, and `thread.terminate()` halts the worker before +// buffered chunks are ever posted, losing output. An empty write's callback +// only fires after every previously buffered chunk has been acked, so awaiting +// it before signaling completion guarantees the output reached the parent. +// A cheap no-op for forks, where stdio goes through OS pipes. +function flushStdio(): Promise { + const flush = (stream: NodeJS.WriteStream) => + new Promise((resolve) => { + try { + stream.write('', () => resolve(undefined)) + } + catch { + resolve(undefined) + } + }) + return Promise.all([flush(process.stdout), flush(process.stderr)]) +} + let traces!: Traces /** @experimental */ @@ -168,6 +189,8 @@ export function init(worker: Options): void { ) const error = await runPromise + await flushStdio() + send({ type: 'testfileFinished', __vitest_worker_response__, @@ -226,6 +249,8 @@ export function init(worker: Options): void { ) const error = await runPromise + await flushStdio() + send({ type: 'testfileFinished', __vitest_worker_response__, @@ -275,11 +300,15 @@ export function init(worker: Options): void { persistCompileCache() + await flushStdio() + send({ type: 'stopped', error, __vitest_worker_response__ }) } catch (error) { persistCompileCache() + await flushStdio() + send({ type: 'stopped', error: serializeError(error), __vitest_worker_response__ }) } diff --git a/test/browser/fixtures/server-url/vitest.config.ts b/test/browser/fixtures/server-url/vitest.config.ts index 542a5af28..a936694fb 100644 --- a/test/browser/fixtures/server-url/vitest.config.ts +++ b/test/browser/fixtures/server-url/vitest.config.ts @@ -23,7 +23,10 @@ export default defineConfig({ !!process.env.TEST_HTTPS && basicSsl(), ], test: { - api: process.env.TEST_HTTPS ? 51122 : 51133, + // below the OS ephemeral port range (32768+ on Linux): a kernel-assigned + // outbound socket holding the fixed port would make Vite silently bind + // port+1 and fail the exact-port assertions + api: process.env.TEST_HTTPS ? 31122 : 31133, browser: { enabled: true, provider: configuredProvider, diff --git a/test/browser/specs/server-url.test.ts b/test/browser/specs/server-url.test.ts index 63a06f270..87815c33c 100644 --- a/test/browser/specs/server-url.test.ts +++ b/test/browser/specs/server-url.test.ts @@ -13,7 +13,7 @@ test('server-url http', async () => { const url = ctx?.projects[0].vite.resolvedUrls?.local[0] expect(stderr).toBe('') expect.assert(url) - expect(new URL(url).port).toBe('51133') + expect(new URL(url).port).toBe('31133') }) test('server-url https', async () => { @@ -25,6 +25,6 @@ test('server-url https', async () => { expect(stderr).toBe('') const url = ctx?.projects[0].vite.resolvedUrls?.local[0] expect.assert(url) - expect(new URL(url).port).toBe('51122') + expect(new URL(url).port).toBe('31122') expect(stdout).toReportSummaryTestFiles({ passed: instances.length }) }) diff --git a/test/e2e/test/concurrent.test.ts b/test/e2e/test/concurrent.test.ts index dbbb19cfa..8780e5e5d 100644 --- a/test/e2e/test/concurrent.test.ts +++ b/test/e2e/test/concurrent.test.ts @@ -10,7 +10,14 @@ import { runInlineTests } from '../../test-utils' // <- * // <------ -const deadlockSource = ` +// In the deadlocking variant "c" resolves the deadlock only after "b" reported +// its timeout: whether a deadlocked test is reported as timed out depends on +// its own elapsed time the moment the deadlock resolves, and "b" starts its +// clock a few event-loop turns after "a", so an unconditional resolve can +// release "b" while it is still within its own budget. The passing variant +// must not gate: nothing fails there, so the gate would never open. +function deadlockSource(gateOnTimeout: boolean) { + return ` import { describe, expect, test } from 'vitest' import { createDefer } from '@vitest/utils/helpers' @@ -20,6 +27,7 @@ describe.concurrent('wrapper', () => { createDefer(), createDefer(), ] + const bTimedOut = createDefer() test('a', async () => { expect(1).toBe(1) @@ -27,7 +35,10 @@ describe.concurrent('wrapper', () => { await defers[2] }) - test('b', async () => { + test('b', async ({ onTestFailed }) => { + onTestFailed(() => { + bTimedOut.resolve() + }) expect(1).toBe(1) await defers[0] defers[1].resolve() @@ -37,14 +48,16 @@ describe.concurrent('wrapper', () => { test('c', async () => { expect(1).toBe(1) await defers[1] + ${gateOnTimeout ? 'await bTimedOut' : ''} defers[2].resolve() }) }) ` +} test('deadlocks with insufficient maxConcurrency', async () => { const { errorTree } = await runInlineTests({ - 'basic.test.ts': deadlockSource, + 'basic.test.ts': deadlockSource(true), }, { maxConcurrency: 2, testTimeout: 500, @@ -74,7 +87,7 @@ test('deadlocks with insufficient maxConcurrency', async () => { test('passes when maxConcurrency is high enough', async () => { const { stderr, errorTree } = await runInlineTests({ - 'basic.test.ts': deadlockSource, + 'basic.test.ts': deadlockSource(false), }, { maxConcurrency: 3, }) @@ -93,7 +106,8 @@ test('passes when maxConcurrency is high enough', async () => { `) }) -const suiteDeadlockSource = ` +function suiteDeadlockSource(gateOnTimeout: boolean) { + return ` import { describe, expect, test } from 'vitest' import { createDefer } from '@vitest/utils/helpers' @@ -103,6 +117,7 @@ describe.concurrent('wrapper', () => { createDefer(), createDefer(), ] + const bTimedOut = createDefer() describe('1st suite', () => { test('a', async () => { @@ -111,7 +126,10 @@ describe.concurrent('wrapper', () => { await defers[2] }) - test('b', async () => { + test('b', async ({ onTestFailed }) => { + onTestFailed(() => { + bTimedOut.resolve() + }) expect(1).toBe(1) await defers[0] defers[1].resolve() @@ -123,15 +141,17 @@ describe.concurrent('wrapper', () => { test('c', async () => { expect(1).toBe(1) await defers[1] + ${gateOnTimeout ? 'await bTimedOut' : ''} defers[2].resolve() }) }) }) ` +} test('suite deadlocks with insufficient maxConcurrency', async () => { const { errorTree } = await runInlineTests({ - 'basic.test.ts': suiteDeadlockSource, + 'basic.test.ts': suiteDeadlockSource(true), }, { maxConcurrency: 2, testTimeout: 500, @@ -162,7 +182,7 @@ test('suite deadlocks with insufficient maxConcurrency', async () => { test('suite passes when maxConcurrency is high enough', async () => { const { stderr, errorTree } = await runInlineTests({ - 'basic.test.ts': suiteDeadlockSource, + 'basic.test.ts': suiteDeadlockSource(false), }, { maxConcurrency: 3, }) diff --git a/test/e2e/test/reporters/import-durations.test.ts b/test/e2e/test/reporters/import-durations.test.ts index 1403a76b7..6a30a9a91 100644 --- a/test/e2e/test/reporters/import-durations.test.ts +++ b/test/e2e/test/reporters/import-durations.test.ts @@ -64,19 +64,21 @@ describe('import durations', () => { }, 40000) it('should handle tests with no imports gracefully', async () => { - const { exitCode, ctx } = await runVitest({ + const { ctx } = await runVitest({ root, include: ['**/ok.test.ts'], experimental: { importDurations: { limit: 10 } }, }) - expect(exitCode).toBe(0) - const capturedFiles = ctx!.state.getFiles() expect(capturedFiles).toHaveLength(1) const file = capturedFiles[0] + // assert on the run's own state, not `exitCode`: `process.exitCode` is + // process-global, and with `isolate: false` a stray unhandled rejection + // leaked by an earlier test file in this worker flips it to 1 + expect(file.result?.state).toBe('pass') expect(file.importDurations).toBeDefined() expect(file.importDurations?.[file.filepath].totalTime).toBeGreaterThanOrEqual(0) expect(file.importDurations?.[file.filepath].selfTime).toBeGreaterThanOrEqual(0) @@ -172,7 +174,7 @@ describe('import durations', () => { it('should fail when failOnDanger is enabled and threshold exceeded', async () => { // With default danger threshold (500ms), should NOT fail (imports are ~75ms) - const { exitCode: exitCodeDefault, stderr: stderrDefault } = await runVitest({ + const { ctx: ctxDefault, stderr: stderrDefault } = await runVitest({ root, include: ['**/import-durations.test.ts'], experimental: { @@ -182,7 +184,9 @@ describe('import durations', () => { }, }) - expect(exitCodeDefault).toBe(0) + // see "should handle tests with no imports gracefully" for why `exitCode` + // is not asserted here + expect(ctxDefault!.state.getFiles()[0]?.result?.state).toBe('pass') expect(stderrDefault).not.toContain('exceeded the danger threshold') // With lower danger threshold (50ms), should fail (imports are ~75ms > 50ms) diff --git a/test/e2e/test/server-url.test.ts b/test/e2e/test/server-url.test.ts index 6e6c09272..5b72a5f69 100644 --- a/test/e2e/test/server-url.test.ts +++ b/test/e2e/test/server-url.test.ts @@ -3,8 +3,13 @@ import { expect, it } from 'vitest' import { runInlineTests } from '../../test-utils' +// `api: true` must resolve to the default port, but which port Vite actually +// binds is not asserted: 51204 sits inside the OS ephemeral port range, and +// when an unrelated outbound socket happens to hold it, Vite silently binds +// port+1 (`strictPort` is off and the "Port is in use" notice is info-level, +// below the logger's `warn` level) it('api server-url http', async () => { - const { stdout, stderr } = await runInlineTests( + const { stdout, stderr, ctx } = await runInlineTests( { 'basic.test.js': `test("basic")` }, { api: true, @@ -12,12 +17,13 @@ it('api server-url http', async () => { }, ) expect(stderr).toBe('') - expect(stdout).toContain('API started at http://localhost:51204/') + expect(ctx!.config.api.port).toBe(51204) + expect(stdout).toMatch(/API started at http:\/\/localhost:\d+\//) expect(stdout).toContain('Test Files 1 skipped') }) it('api server-url https', async () => { - const { stdout, stderr } = await runInlineTests( + const { stdout, stderr, ctx } = await runInlineTests( { 'basic.test.js': `test("basic")` }, { api: true, @@ -28,7 +34,8 @@ it('api server-url https', async () => { }, ) expect(stderr).toBe('') - expect(stdout).toContain('API started at https://localhost:51204/') + expect(ctx!.config.api.port).toBe(51204) + expect(stdout).toMatch(/API started at https:\/\/localhost:\d+\//) expect(stdout).toContain('Test Files 1 skipped') }) -- 2.51.2