diff --git a/common/changes/@rushstack/heft-jest-plugin/user-yverupaka-jest-worker-lifecycle-pr_2026-11-13-19-03-49.json b/common/changes/@rushstack/heft-jest-plugin/user-yverupaka-jest-worker-lifecycle-pr_2026-11-13-19-03-49.json new file mode 100644 index 00000000000..05f9caf9cf9 --- /dev/null +++ b/common/changes/@rushstack/heft-jest-plugin/user-yverupaka-jest-worker-lifecycle-pr_2026-11-13-19-03-49.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "comment": "Add a standalone sequential Jest worker lifecycle supervisor without enabling managed execution.", + "type": "minor", + "packageName": "@rushstack/heft-jest-plugin" + } + ], + "packageName": "@rushstack/heft-jest-plugin", + "email": "59129034+yashwanth195@users.noreply.github.com" +} \ No newline at end of file diff --git a/common/config/rush/nonbrowser-approved-packages.json b/common/config/rush/nonbrowser-approved-packages.json index 853d3fdc370..80b44caf67a 100644 --- a/common/config/rush/nonbrowser-approved-packages.json +++ b/common/config/rush/nonbrowser-approved-packages.json @@ -830,6 +830,10 @@ "name": "jest-watch-select-projects", "allowedCategories": [ "libraries" ] }, + { + "name": "jest-worker", + "allowedCategories": [ "libraries" ] + }, { "name": "jju", "allowedCategories": [ "libraries" ] diff --git a/common/config/subspaces/build-tests-subspace/pnpm-lock.yaml b/common/config/subspaces/build-tests-subspace/pnpm-lock.yaml index 08f760e7c08..f393ef9c316 100644 --- a/common/config/subspaces/build-tests-subspace/pnpm-lock.yaml +++ b/common/config/subspaces/build-tests-subspace/pnpm-lock.yaml @@ -4837,6 +4837,7 @@ snapshots: jest-config: 30.3.0(@types/node@20.17.19) jest-resolve: 30.3.0 jest-snapshot: 30.3.0 + jest-worker: 30.3.0 optionalDependencies: '@types/jest': 30.0.0 jest-environment-node: 30.3.0 diff --git a/common/config/subspaces/build-tests-subspace/repo-state.json b/common/config/subspaces/build-tests-subspace/repo-state.json index 1449961bcde..d3c6bb3eb5c 100644 --- a/common/config/subspaces/build-tests-subspace/repo-state.json +++ b/common/config/subspaces/build-tests-subspace/repo-state.json @@ -1,6 +1,6 @@ // DO NOT MODIFY THIS FILE MANUALLY BUT DO COMMIT IT. It is generated and used by Rush. { - "pnpmShrinkwrapHash": "18a657d83020df41f6602b7c9087b7253cfee8e0", + "pnpmShrinkwrapHash": "f7e1690c6e376aa71bbfabe99fe991613925b896", "preferredVersionsHash": "550b4cee0bef4e97db6c6aad726df5149d20e7d9", - "packageJsonInjectedDependenciesHash": "293debc6ed16567f5567a7748229a73be6647cc6" + "packageJsonInjectedDependenciesHash": "be3a016733c367a2e9e3e067c1d5144ab08ca94a" } diff --git a/common/config/subspaces/default/pnpm-lock.yaml b/common/config/subspaces/default/pnpm-lock.yaml index 13b7ce39782..f326d8e5d6f 100644 --- a/common/config/subspaces/default/pnpm-lock.yaml +++ b/common/config/subspaces/default/pnpm-lock.yaml @@ -3325,6 +3325,9 @@ importers: jest-snapshot: specifier: ~30.3.0 version: 30.3.0 + jest-worker: + specifier: ~30.3.0 + version: 30.3.0 devDependencies: '@jest/types': specifier: 30.3.0 diff --git a/heft-plugins/heft-jest-plugin/README.md b/heft-plugins/heft-jest-plugin/README.md index 25f82e5ef62..16fa0e45202 100644 --- a/heft-plugins/heft-jest-plugin/README.md +++ b/heft-plugins/heft-jest-plugin/README.md @@ -2,6 +2,11 @@ This is a Heft plugin for running Jest. +`lib/JestManagedWorkerQueue` provides lifecycle supervision for a single sequential `jest-worker`. +Await tasks through `waitForTaskAsync(task)` so failure observation ends when each task settles. +The class documents the remaining contract: no retries, idle-recycle readiness between tasks, and +explicit final-task/shutdown handling. Ordinary Jest execution does not use it or enable managed sharding. + ## Links - [CHANGELOG.md]( diff --git a/heft-plugins/heft-jest-plugin/package.json b/heft-plugins/heft-jest-plugin/package.json index 53c493bc2ac..4141f666195 100644 --- a/heft-plugins/heft-jest-plugin/package.json +++ b/heft-plugins/heft-jest-plugin/package.json @@ -45,7 +45,8 @@ "@rushstack/terminal": "workspace:*", "jest-config": "~30.3.0", "jest-resolve": "~30.3.0", - "jest-snapshot": "~30.3.0" + "jest-snapshot": "~30.3.0", + "jest-worker": "~30.3.0" }, "devDependencies": { "@jest/types": "30.3.0", diff --git a/heft-plugins/heft-jest-plugin/src/JestManagedWorkerQueue.ts b/heft-plugins/heft-jest-plugin/src/JestManagedWorkerQueue.ts new file mode 100644 index 00000000000..c92c606f807 --- /dev/null +++ b/heft-plugins/heft-jest-plugin/src/JestManagedWorkerQueue.ts @@ -0,0 +1,271 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { EventEmitter } from 'node:events'; +import { totalmem } from 'node:os'; + +import { FifoQueue, type WorkerPoolInterface } from 'jest-worker'; + +type WorkerHandle = ReturnType[number]; +type QueuedTask = Parameters[0]; + +/** + * Supervises one sequential Jest worker through public task callbacks and lifecycle handles. + * + * @remarks + * Construct Worker with numWorkers=1, maxRetries=0, this taskQueue and the same idleMemoryLimit. + * Dispatch the first task synchronously after construction to observe the initial handle. + * Await each task through waitForTaskAsync, then await waitForIdleAsync before dispatching another task. + * Mark the final task before enqueueing it. Check throwIfFailed before beginShutdown, worker.end + * and dispose (in finally); tolerate end's forceExited only when replacedAfterFinalTask is true. + * This queue does not schedule concurrent work or implement caller cancellation. + */ +export class JestManagedWorkerQueue extends FifoQueue { + readonly #idleMemoryLimit: number | undefined; + readonly #failureWaiters: Set<(error: Error) => void> = new Set(); + #error: Error | undefined; + #worker: WorkerHandle | undefined; + #unsubscribe: (() => void) | undefined; + #idle: Promise = Promise.resolve(); + #resolveReady: (() => void) | undefined; + #recycling: boolean = false; + #finalTask: boolean = false; + #finalTaskCompleted: boolean = false; + #classifyingFinalRecycle: boolean = false; + #replacedAfterFinalTask: boolean = false; + #resolveFinalState: ((state: WorkerHandle['state']) => void) | undefined; + #ending: boolean = false; + + public constructor(idleMemoryLimit: number | undefined) { + super(); + this.#idleMemoryLimit = + idleMemoryLimit === undefined + ? undefined + : Math.floor( + idleMemoryLimit > 0 && idleMemoryLimit <= 1 ? totalmem() * idleMemoryLimit : idleMemoryLimit + ); + } + + public override enqueue(task: QueuedTask, workerId?: number): void { + const isFinalTask: boolean = this.#finalTask; + super.enqueue( + { + ...task, + onStart: (worker) => { + if (this.#worker !== worker) { + if (this.#worker) { + this.#fail(new Error('Managed Jest worker handle was replaced unexpectedly.')); + // The old handle was stopped by fail; the newly supplied handle is also owned. + worker.forceExit(); + } + this.#unsubscribe?.(); + this.#worker = worker; + // WorkerOptions.on declares "state-change"; built-in handles inherit EventEmitter. + if (!(worker instanceof EventEmitter)) { + this.#fail(new Error('Jest worker does not expose its lifecycle events.')); + } else { + this.#unsubscribe = this.#subscribeToStateChanges(worker); + this.#observeState(worker.state); + } + } + task.onStart(worker); + }, + onEnd: (error, result) => { + if (!error && isFinalTask) { + this.#finalTaskCompleted = true; + const worker: WorkerHandle | undefined = this.#worker; + const limit: number | undefined = this.#idleMemoryLimit; + if (worker && limit !== undefined) { + this.#classifyingFinalRecycle = true; + const finalState: Promise = new Promise((resolve) => { + this.#resolveFinalState = resolve; + }); + this.#idle = this.#classifyFinalRecycleAsync(worker, limit, finalState) + .catch((idleError: Error) => { + this.#fail(idleError); + }) + .finally(() => { + this.#resolveFinalState = undefined; + this.#classifyingFinalRecycle = false; + this.#observeState(worker.state); + }); + this.#idle.catch((idleError: Error) => this.#fail(idleError)); + } + } else if (!error && !this.#ending && this.#worker) { + const worker: WorkerHandle = this.#worker; + const limit: number | undefined = this.#idleMemoryLimit; + if (limit !== undefined) { + this.#idle = this.#waitForIdleAsync(worker, limit); + this.#idle.catch((idleError: Error) => this.#fail(idleError)); + } + } + task.onEnd(error, result); + } + }, + workerId + ); + } + + /** + * Observes lifecycle failures only until this task settles, without retaining completed task results. + * Rejects immediately if the queue has already failed; otherwise forwards the first observed outcome. + * This does not cancel the supplied task. + */ + public waitForTaskAsync(task: Promise): Promise { + return new Promise((resolve, reject) => { + if (this.#error) { + reject(this.#error); + } else { + this.#failureWaiters.add(reject); + } + // Always observe task rejection, even if a lifecycle failure has already won. + task.then( + (result) => { + this.#failureWaiters.delete(reject); + resolve(result); + }, + (error: unknown) => { + this.#failureWaiters.delete(reject); + reject(error); + } + ); + }); + } + + public async waitForIdleAsync(): Promise { + await this.waitForTaskAsync(this.#idle); + this.throwIfFailed(); + } + + public throwIfFailed(): void { + if (this.#error) throw this.#error; + } + + public beginShutdown(): void { + this.#ending = true; + } + + public get replacedAfterFinalTask(): boolean { + return this.#replacedAfterFinalTask; + } + + public markFinalTask(): void { + if (this.#finalTask || this.#ending) { + throw new Error('Managed Jest final worker task was already marked.'); + } + this.#finalTask = true; + } + + public dispose(): void { + this.#unsubscribe?.(); + } + + // Create long-lived listeners outside the enqueue closure so they do not retain its first task. + #subscribeToStateChanges(worker: WorkerHandle & EventEmitter): () => void { + const onStateChange = (state: WorkerHandle['state']): void => { + // Memory resolves before RESTARTING; zero-limit processes restart before onEnd. + // Defer classification until the memory evidence and final-task state are installed. + queueMicrotask(() => this.#observeState(state)); + }; + worker.on('state-change', onStateChange); + return () => worker.off('state-change', onStateChange); + } + + async #waitForIdleAsync(worker: WorkerHandle, limit: number): Promise { + if (limit === 0) { + // Processes support unconditional recycling; threads currently treat zero as disabled. + this.#recycling = worker.state === 'restarting'; + } else { + const memory: number | null = await worker.getMemoryUsage(); + if (memory === null) throw new Error('Jest worker did not report its idle memory usage.'); + this.#recycling = memory > limit; + } + if (this.#recycling) { + await new Promise((resolve) => { + this.#resolveReady = resolve; + this.#observeState(worker.state); + }); + // "ok" means spawned, not bootstrapped. A memory response proves message-loop readiness + // without running another task (which would itself trigger idle-memory recycling). + if ((await worker.getMemoryUsage()) === null) { + throw new Error('Recycled Jest worker did not acknowledge readiness.'); + } + } + } + + async #classifyFinalRecycleAsync( + worker: WorkerHandle, + limit: number, + finalState: Promise + ): Promise { + if (limit === 0) { + this.#recycling = worker.state === 'restarting'; + } else { + const result: + | { memory: number | null; state?: undefined } + | { memory?: undefined; state: WorkerHandle['state'] } = await Promise.race([ + worker.getMemoryUsage().then((memory) => ({ memory })), + finalState.then((state) => ({ state })) + ]); + if (result.state !== undefined) { + throw new Error(`Managed Jest worker stopped before reporting final memory usage (${result.state}).`); + } + const memory: number | null = result.memory; + if (memory === null) throw new Error('Jest worker did not report its idle memory usage.'); + this.#recycling = memory > limit; + } + if (this.#recycling && (worker.state === 'restarting' || worker.state === 'starting')) { + this.#replacedAfterFinalTask = true; + } + } + + #observeState(state: WorkerHandle['state']): void { + if ( + this.#finalTaskCompleted && + (state === 'restarting' || + state === 'starting' || + state === 'oom' || + state === 'shutting-down' || + state === 'shut-down') + ) { + this.#resolveFinalState?.(state); + } + if (this.#finalTaskCompleted && this.#recycling && (state === 'restarting' || state === 'starting')) { + this.#replacedAfterFinalTask = true; + } + if (this.#ending || this.#classifyingFinalRecycle) return; + switch (state) { + case 'ok': + if (this.#recycling) { + this.#recycling = false; + this.#resolveReady?.(); + this.#resolveReady = undefined; + } + break; + case 'restarting': + case 'starting': + if (!this.#recycling) { + this.#fail(new Error('Managed Jest worker restarted unexpectedly.')); + } + break; + case 'oom': + case 'shutting-down': + case 'shut-down': + this.#fail(new Error(`Managed Jest worker stopped unexpectedly (${state}).`)); + break; + } + } + + #fail(error: Error): void { + if (this.#error) return; + this.#error = error; + for (const reject of this.#failureWaiters) { + reject(error); + } + this.#failureWaiters.clear(); + // maxRetries=0 prevents replay, but Jest may already have spawned a replacement. + if (this.#worker && this.#worker.state !== 'shut-down' && this.#worker.state !== 'shutting-down') { + this.#worker.forceExit(); + } + } +} diff --git a/heft-plugins/heft-jest-plugin/src/test/JestManagedWorkerQueue.integration.test.ts b/heft-plugins/heft-jest-plugin/src/test/JestManagedWorkerQueue.integration.test.ts new file mode 100644 index 00000000000..54a24fa3594 --- /dev/null +++ b/heft-plugins/heft-jest-plugin/src/test/JestManagedWorkerQueue.integration.test.ts @@ -0,0 +1,154 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as fs from 'node:fs/promises'; +import * as path from 'node:path'; +import { Worker, type JestWorkerFarm } from 'jest-worker'; + +import { JestManagedWorkerQueue } from '../JestManagedWorkerQueue'; +import type * as fixture from './fixtures/managedWorker'; + +type FixtureWorker = JestWorkerFarm; +type WorkerIdentity = fixture.IWorkerIdentity; + +const fixturePath: string = path.resolve( + __dirname, + '..', + '..', + 'lib-commonjs', + 'test', + 'fixtures', + 'managedWorker.js' +); +const tempPath: string = path.resolve(__dirname, '..', '..', 'temp', 'managed-workers'); + +describe.each([false, true])('Jest managed worker lifecycle (threads=%s)', (enableWorkerThreads) => { + let folder: string; + let logPath: string; + let worker: FixtureWorker | undefined; + let queue: JestManagedWorkerQueue; + + beforeEach(async () => { + await fs.mkdir(tempPath, { recursive: true }); + folder = await fs.mkdtemp(path.join(tempPath, 'worker-')); + logPath = path.join(folder, 'events.log'); + }); + + afterEach(async () => { + try { + if (worker) { + await endAsync(); + } + } finally { + await fs.rm(folder, { recursive: true, force: true }); + } + }); + + function createWorker(idleMemoryLimit?: number, env?: NodeJS.ProcessEnv): FixtureWorker { + queue = new JestManagedWorkerQueue(idleMemoryLimit); + worker = new Worker(fixturePath, { + numWorkers: 1, + maxRetries: 0, + idleMemoryLimit, + enableWorkerThreads, + taskQueue: queue, + exposedMethods: ['run'], + forkOptions: { + serialization: 'json', + // Preload before Jest's bootstrap to exercise initial-handle supervision and readiness. + execArgv: ['--require', fixturePath], + env: { ...process.env, MANAGED_WORKER_LOG: logPath, ...env } + } + }) as FixtureWorker; + return worker; + } + + async function endAsync(): Promise>> { + const active: FixtureWorker = worker!; + worker = undefined; + queue.beginShutdown(); + try { + return await active.end(); + } finally { + queue.dispose(); + } + } + + async function finishAsync(): Promise { + queue.throwIfFailed(); + const result: Awaited> = await endAsync(); + expect(result.forceExited && !queue.replacedAfterFinalTask).toBe(false); + } + + it.each([undefined, 1024, 0])( + 'reuses or recycles between tasks without replay (idleMemoryLimit=%s)', + async (limit) => { + const farm: FixtureWorker = createWorker(limit); + const identities: WorkerIdentity[] = []; + for (let index: number = 0; index < 3; index++) { + if (index === 2) queue.markFinalTask(); + identities.push(await queue.waitForTaskAsync(farm.run(`task-${index}`))); + await queue.waitForIdleAsync(); + } + const recycles: boolean = limit === 1024 || (limit === 0 && !enableWorkerThreads); + expect(new Set(identities.map(({ pid, threadId }) => `${pid}:${threadId}`)).size).toBe( + recycles ? 3 : 1 + ); + expect(identities.every((identity) => identity.isMainThread === !enableWorkerThreads)).toBe(true); + expect(queue.replacedAfterFinalTask).toBe(recycles); + await finishAsync(); + const events: string = await fs.readFile(logPath, 'utf8'); + expect(events.split('\n').filter((line) => line.startsWith('run '))).toHaveLength(3); + }, + 30000 + ); + + it('waits for a slow replacement bootstrap before another task or graceful shutdown', async () => { + const farm: FixtureWorker = createWorker(1024, { MANAGED_WORKER_BOOT_DELAY: '1' }); + const first: WorkerIdentity = await queue.waitForTaskAsync(farm.run('first')); + await queue.waitForIdleAsync(); + const beforeNext: string = await fs.readFile(logPath, 'utf8'); + expect(beforeNext.split('\n').filter((line) => line.startsWith('ready '))).toHaveLength(2); + const second: WorkerIdentity = await queue.waitForTaskAsync(farm.run('second')); + expect(second).not.toEqual(first); + await queue.waitForIdleAsync(); + const beforeEnd: string = await fs.readFile(logPath, 'utf8'); + expect(beforeEnd.split('\n').filter((line) => line.startsWith('ready '))).toHaveLength(3); + expect(beforeEnd.split('\n').filter((line) => line.startsWith('run '))).toHaveLength(2); + // Deliberately do not mark a final task: readiness must make normal shutdown graceful. + expect(queue.replacedAfterFinalTask).toBe(false); + await finishAsync(); + }, 30000); + + it.each([0, 7])( + 'surfaces exit %s without replaying in-flight work', + async (exitCode) => { + const farm: FixtureWorker = createWorker(); + await expect(queue.waitForTaskAsync(farm.run('crash', exitCode))).rejects.toThrow(); + await expect(queue.waitForIdleAsync()).rejects.toThrow( + /Managed Jest worker (stopped|restarted) unexpectedly/ + ); + expect(() => queue.throwIfFailed()).toThrow(); + expect(queue.replacedAfterFinalTask).toBe(false); + await endAsync(); + const events: string = await fs.readFile(logPath, 'utf8'); + expect(events.split('\n').filter((line) => line.startsWith('run crash '))).toHaveLength(1); + }, + 30000 + ); + + it.each([0, 7])( + 'supervises initial bootstrap exit %s before the first task starts', + async (exitCode) => { + const farm: FixtureWorker = createWorker(undefined, { MANAGED_WORKER_BOOT_EXIT: String(exitCode) }); + await expect(queue.waitForTaskAsync(farm.run('never-started'))).rejects.toThrow(); + await expect(queue.waitForIdleAsync()).rejects.toThrow( + /Managed Jest worker (stopped|restarted) unexpectedly/ + ); + await endAsync(); + const events: string = await fs.readFile(logPath, 'utf8'); + expect(events).not.toContain('run '); + }, + 30000 + ); +}); diff --git a/heft-plugins/heft-jest-plugin/src/test/JestManagedWorkerQueue.test.ts b/heft-plugins/heft-jest-plugin/src/test/JestManagedWorkerQueue.test.ts new file mode 100644 index 00000000000..fddffa241f4 --- /dev/null +++ b/heft-plugins/heft-jest-plugin/src/test/JestManagedWorkerQueue.test.ts @@ -0,0 +1,302 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +jest.mock('node:os', () => ({ totalmem: () => 1000 })); + +import { EventEmitter } from 'node:events'; +import * as v8 from 'node:v8'; +import * as vm from 'node:vm'; +import type { FifoQueue, WorkerPoolInterface } from 'jest-worker'; + +import { JestManagedWorkerQueue } from '../JestManagedWorkerQueue'; + +type WorkerHandle = ReturnType[number]; +type QueuedTask = Parameters[0]; +type MockWorkerHandle = jest.Mocked & EventEmitter & { state: WorkerHandle['state'] }; + +function createHandle(): MockWorkerHandle { + return Object.assign(new EventEmitter(), { + state: 'ok' as WorkerHandle['state'], + send: jest.fn(), + waitForExit: jest.fn(), + forceExit: jest.fn(), + getWorkerId: jest.fn(), + getStderr: jest.fn(), + getStdout: jest.fn(), + getWorkerSystemId: jest.fn(), + getMemoryUsage: jest.fn, []>().mockResolvedValue(100), + isWorkerRunning: jest.fn(), + waitForWorkerReady: jest.fn() + }); +} + +function setState(worker: ReturnType, state: `${WorkerHandle['state']}`): void { + worker.state = state as WorkerHandle['state']; + worker.emit('state-change', worker.state); +} + +function startTask( + queue: JestManagedWorkerQueue, + worker: WorkerHandle, + onEnd: QueuedTask['onEnd'] = jest.fn() +): QueuedTask { + queue.enqueue({ + request: [1, false, 'run', []], + onStart: jest.fn(), + onEnd, + onCustomMessage: jest.fn() + }); + const task: QueuedTask = queue.dequeue(0)!; + task.onStart(worker); + return task; +} + +function getGarbageCollection(): () => void { + const exposed: unknown = (globalThis as { gc?: unknown }).gc; + if (typeof exposed === 'function') { + return exposed as () => void; + } + v8.setFlagsFromString('--expose-gc'); + try { + return vm.runInNewContext('gc'); + } finally { + // Do not expose gc to subsequent Jest contexts sharing this worker. + v8.setFlagsFromString('--no-expose-gc'); + } +} + +describe(JestManagedWorkerQueue.name, () => { + let queue: JestManagedWorkerQueue; + let worker: ReturnType; + + beforeEach(() => { + queue = new JestManagedWorkerQueue(undefined); + worker = createHandle(); + }); + + afterEach(() => { + queue.beginShutdown(); + queue.dispose(); + expect(worker.listenerCount('state-change')).toBe(0); + }); + + it('releases completed task results while the queue remains alive', async () => { + const collectGarbage: () => void = getGarbageCollection(); + const observed: WeakRef[] = []; + const control: WeakRef[] = []; + + async function completeAsync(observeFailure: boolean): Promise { + let resolveTask!: (result: unknown) => void; + const pending: Promise = new Promise((resolve) => { + resolveTask = resolve; + }); + const task: QueuedTask = startTask(queue, worker, (error, result) => resolveTask(result)); + const awaited: Promise = observeFailure ? queue.waitForTaskAsync(pending) : pending; + const result: object = { testResults: [{ title: 'completed suite' }] }; + (observeFailure ? observed : control).push(new WeakRef(result)); + task.onEnd(null, result); + await awaited; + await queue.waitForIdleAsync(); + } + + for (let index: number = 0; index < 12; index++) { + await completeAsync(true); + await completeAsync(false); + } + // WeakRef targets stay alive through the current job, including after deref(). + for (let index: number = 0; index < 2; index++) { + await new Promise((resolve) => setImmediate(resolve)); + collectGarbage(); + } + expect(control.filter((reference) => reference.deref() !== undefined)).toHaveLength(0); + expect(observed.filter((reference) => reference.deref() !== undefined)).toHaveLength(0); + queue.throwIfFailed(); + }); + + it('preserves task result and rejection identity without failing the worker', async () => { + const result: object = { suite: 'passed' }; + const error: Error = new Error('suite failed'); + await expect(queue.waitForTaskAsync(Promise.resolve(result))).resolves.toBe(result); + await expect(queue.waitForTaskAsync(Promise.reject(error))).rejects.toBe(error); + await expect(queue.waitForIdleAsync()).resolves.toBeUndefined(); + expect(() => queue.throwIfFailed()).not.toThrow(); + expect(worker.forceExit).not.toHaveBeenCalled(); + }); + + it('interrupts pending tasks and idle waits while consuming late task rejections', async () => { + queue = new JestManagedWorkerQueue(100); + worker.getMemoryUsage.mockReturnValue(new Promise(() => {})); + startTask(queue, worker).onEnd(null, 'first'); + let rejectTask!: (error: Error) => void; + const pending: Promise = new Promise((resolve, reject) => { + rejectTask = reject; + }); + const taskFailure: Promise = queue.waitForTaskAsync(pending).catch((error: Error) => error); + const idleFailure: Promise = queue.waitForIdleAsync().catch((error: Error) => error); + setState(worker, 'oom'); + const failure: Error = await taskFailure; + expect(failure.message).toContain('stopped unexpectedly (oom)'); + expect(await idleFailure).toBe(failure); + rejectTask(new Error('late task rejection')); + await expect(queue.waitForTaskAsync(Promise.reject(new Error('another task rejection')))).rejects.toBe( + failure + ); + await expect(queue.waitForTaskAsync(Promise.resolve('late result'))).rejects.toBe(failure); + setState(worker, 'shut-down'); + await new Promise((resolve) => setImmediate(resolve)); + expect(() => queue.throwIfFailed()).toThrow(failure); + expect(worker.forceExit).toHaveBeenCalledTimes(1); + }); + + it.each([false, true])( + 'retains lifecycle failure when racing task completion (taskFirst=%s)', + async (taskFirst) => { + startTask(queue, worker); + const result: object = { suite: 'completed' }; + let resolveTask!: (value: object) => void; + const pending: Promise = new Promise((resolve) => { + resolveTask = resolve; + }); + const observed: Promise = queue.waitForTaskAsync(pending); + const outcome: Promise = taskFirst + ? expect(observed).resolves.toBe(result) + : expect(observed).rejects.toThrow('stopped unexpectedly (oom)'); + if (taskFirst) { + resolveTask(result); + await Promise.resolve(); + setState(worker, 'oom'); + } else { + setState(worker, 'oom'); + await Promise.resolve(); + resolveTask(result); + } + await outcome; + expect(() => queue.throwIfFailed()).toThrow('stopped unexpectedly (oom)'); + expect(worker.forceExit).toHaveBeenCalledTimes(1); + } + ); + + it.each([ + [0.1, 100], + [1, 1000], + [100.9, 100] + ])('normalizes limit %s and waits for replacement readiness only above %s bytes', async (limit, bytes) => { + queue = new JestManagedWorkerQueue(limit); + worker.getMemoryUsage.mockResolvedValueOnce(bytes); + startTask(queue, worker).onEnd(null, 'first'); + await queue.waitForIdleAsync(); + expect(worker.getMemoryUsage).toHaveBeenCalledTimes(1); + + let acknowledgeReady!: (memory: number) => void; + const ready: Promise = new Promise((resolve) => { + acknowledgeReady = resolve; + }); + worker.getMemoryUsage.mockResolvedValueOnce(bytes + 1).mockReturnValueOnce(ready); + startTask(queue, worker).onEnd(null, 'second'); + setState(worker, 'restarting'); + await Promise.resolve(); + setState(worker, 'starting'); + setState(worker, 'ok'); + let idle: boolean = false; + const pending: Promise = queue.waitForIdleAsync().then(() => { + idle = true; + }); + // Let the lifecycle and memory continuations run, but do not acknowledge the replacement. + await new Promise((resolve) => setImmediate(resolve)); + expect(worker.getMemoryUsage).toHaveBeenCalledTimes(3); + expect(idle).toBe(false); + acknowledgeReady(bytes); + await pending; + expect(worker.forceExit).not.toHaveBeenCalled(); + }); + + it.each(['starting', 'restarting', 'oom', 'shutting-down', 'shut-down'] as const)( + 'retains unexpected %s failures and stops the owned worker', + async (state) => { + startTask(queue, worker); + setState(worker, state); + const error: Error = await queue + .waitForTaskAsync(new Promise(() => {})) + .catch((failure: Error) => failure); + expect(error).toBeInstanceOf(Error); + expect(() => queue.throwIfFailed()).toThrow(error); + await expect(queue.waitForIdleAsync()).rejects.toBe(error); + expect(worker.forceExit).toHaveBeenCalledTimes( + state === 'shutting-down' || state === 'shut-down' ? 0 : 1 + ); + setState(worker, 'ok'); + expect(() => queue.throwIfFailed()).toThrow(error); + } + ); + + it('observes an already stopped initial handle without waiting for another event', async () => { + setState(worker, 'shut-down'); + startTask(queue, worker); + await expect(queue.waitForIdleAsync()).rejects.toThrow('stopped unexpectedly (shut-down)'); + }); + + it('stops both owned handles on unexpected handle replacement', async () => { + startTask(queue, worker).onEnd(null, 'first'); + const replacement: ReturnType = createHandle(); + startTask(queue, replacement); + await expect(queue.waitForIdleAsync()).rejects.toThrow('handle was replaced unexpectedly'); + expect(worker.forceExit).toHaveBeenCalledTimes(1); + expect(replacement.forceExit).toHaveBeenCalledTimes(1); + queue.dispose(); + expect(replacement.listenerCount('state-change')).toBe(0); + }); + + it('fails and cleans up when idle memory is unavailable', async () => { + queue = new JestManagedWorkerQueue(100); + worker.getMemoryUsage.mockResolvedValue(null); + startTask(queue, worker).onEnd(null, 'result'); + await expect(queue.waitForIdleAsync()).rejects.toThrow('did not report its idle memory'); + expect(worker.forceExit).toHaveBeenCalledTimes(1); + }); + + it.each(['restarting', 'oom', 'shut-down'] as const)( + 'does not hang or accept a final %s without memory evidence', + async (state) => { + queue = new JestManagedWorkerQueue(100); + worker.getMemoryUsage.mockReturnValue(new Promise(() => {})); + queue.markFinalTask(); + startTask(queue, worker).onEnd(null, 'result'); + setState(worker, state); + await expect(queue.waitForIdleAsync()).rejects.toThrow( + `stopped before reporting final memory usage (${state})` + ); + expect(queue.replacedAfterFinalTask).toBe(false); + expect(() => queue.throwIfFailed()).toThrow('stopped before reporting final memory'); + } + ); + + it.each([0, 100])('classifies final recycling at limit %s without waiting for readiness', async (limit) => { + queue = new JestManagedWorkerQueue(limit); + worker.getMemoryUsage.mockResolvedValue(101); + queue.markFinalTask(); + const task: QueuedTask = startTask(queue, worker); + if (limit === 0) setState(worker, 'restarting'); + task.onEnd(null, 'result'); + if (limit !== 0) setState(worker, 'restarting'); + await queue.waitForIdleAsync(); + expect(queue.replacedAfterFinalTask).toBe(true); + expect(worker.getMemoryUsage).toHaveBeenCalledTimes(limit === 0 ? 0 : 1); + queue.beginShutdown(); + setState(worker, 'shutting-down'); + setState(worker, 'shut-down'); + await queue.waitForIdleAsync(); + expect(worker.forceExit).not.toHaveBeenCalled(); + }); + + it('does not classify a failed final task as a successful recycle', async () => { + queue = new JestManagedWorkerQueue(100); + queue.markFinalTask(); + const task: QueuedTask = startTask(queue, worker); + const error: Error = new Error('task failed'); + task.onEnd(error, undefined); + setState(worker, 'restarting'); + await expect(queue.waitForIdleAsync()).rejects.toThrow('restarted unexpectedly'); + expect(queue.replacedAfterFinalTask).toBe(false); + expect(worker.getMemoryUsage).not.toHaveBeenCalled(); + }); +}); diff --git a/heft-plugins/heft-jest-plugin/src/test/fixtures/managedWorker.ts b/heft-plugins/heft-jest-plugin/src/test/fixtures/managedWorker.ts new file mode 100644 index 00000000000..c53d871b839 --- /dev/null +++ b/heft-plugins/heft-jest-plugin/src/test/fixtures/managedWorker.ts @@ -0,0 +1,33 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { appendFileSync, existsSync, readFileSync } from 'node:fs'; +import { isMainThread, threadId } from 'node:worker_threads'; + +export interface IWorkerIdentity { + pid: number; + threadId: number; + isMainThread: boolean; +} + +const logPath: string | undefined = process.env.MANAGED_WORKER_LOG; +if (!logPath) throw new Error('Missing fixture log path.'); +const firstBootstrap: boolean = !existsSync(logPath); +const identity: IWorkerIdentity = { pid: process.pid, threadId, isMainThread }; +const id: string = `${identity.pid}:${identity.threadId}`; +appendFileSync(logPath, `boot ${id}\n`); +if (firstBootstrap && process.env.MANAGED_WORKER_BOOT_EXIT) { + process.exit(Number(process.env.MANAGED_WORKER_BOOT_EXIT)); +} +if (process.env.MANAGED_WORKER_BOOT_DELAY) { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 750); +} +appendFileSync(logPath, `ready ${id}\n`); + +export function run(label: string, exitCode?: number): IWorkerIdentity { + const firstAttempt: boolean = !readFileSync(logPath!, 'utf8').includes(`run ${label} `); + appendFileSync(logPath!, `run ${label} ${id}\n`); + // A replay would return successfully, so the parent must reject and record exactly one attempt. + if (firstAttempt && exitCode !== undefined) process.exit(exitCode); + return identity; +}