diff --git a/packages/nx/src/command-line/graph/graph.ts b/packages/nx/src/command-line/graph/graph.ts index 12673a06c90..aebff50b277 100644 --- a/packages/nx/src/command-line/graph/graph.ts +++ b/packages/nx/src/command-line/graph/graph.ts @@ -395,7 +395,6 @@ export async function generateGraph( ) ); await output.drain(); - await new Promise((res) => setImmediate(res)); process.exit(0); } @@ -480,7 +479,7 @@ export async function generateGraph( }); process.exit(1); } - await new Promise((res) => setImmediate(res)); + await output.drain(); process.exit(0); } else { const environmentJs = buildEnvironmentJs( diff --git a/packages/nx/src/command-line/release/command-object.ts b/packages/nx/src/command-line/release/command-object.ts index 41a92a7e66b..78418622982 100644 --- a/packages/nx/src/command-line/release/command-object.ts +++ b/packages/nx/src/command-line/release/command-object.ts @@ -1,6 +1,7 @@ import { type Argv, type CommandModule, showHelp } from 'yargs'; import { handleImport } from '../../utils/handle-import'; import { logger } from '../../utils/logger'; +import { output } from '../../utils/output'; import { type OutputStyle, type RunManyOptions, @@ -263,6 +264,7 @@ const releaseCommand: CommandModule = { logger.warn(`\nNOTE: The "dryRun" flag means no changes were made.`); } + await output.drain(); process.exit(result); }, }; @@ -303,6 +305,7 @@ const versionCommand: CommandModule = { logger.warn(`\nNOTE: The "dryRun" flag means no changes were made.`); } + await output.drain(); process.exit(result); }, }; @@ -371,6 +374,7 @@ const changelogCommand: CommandModule = { logger.warn(`\nNOTE: The "dryRun" flag means no changes were made.`); } + await output.drain(); process.exit(result); }, }; @@ -411,6 +415,7 @@ const publishCommand: CommandModule = { logger.warn(`\nNOTE: The "dryRun" flag means no changes were made.`); } + await output.drain(); process.exit(status); }, }; @@ -453,6 +458,7 @@ const planCommand: CommandModule = { logger.warn(`\nNOTE: The "dryRun" flag means no changes were made.`); } + await output.drain(); process.exit(result); }, }; @@ -465,6 +471,7 @@ const planCheckCommand: CommandModule = { handler: async (args) => { const release = await handleImport('./plan-check.js', __dirname); const result = await release.releasePlanCheckCLIHandler(args); + await output.drain(); process.exit(result); }, }; diff --git a/packages/nx/src/command-line/run-many/run-many.ts b/packages/nx/src/command-line/run-many/run-many.ts index 008b1be6bf4..feeaaef5420 100644 --- a/packages/nx/src/command-line/run-many/run-many.ts +++ b/packages/nx/src/command-line/run-many/run-many.ts @@ -76,6 +76,7 @@ export async function runMany( extraTargetDependencies, extraOptions ); + await output.drain(); process.exit(status); } } diff --git a/packages/nx/src/command-line/run/run-one.ts b/packages/nx/src/command-line/run/run-one.ts index edbafc86b7e..3c75ca120b1 100644 --- a/packages/nx/src/command-line/run/run-one.ts +++ b/packages/nx/src/command-line/run/run-one.ts @@ -69,6 +69,7 @@ export async function runOne( }, workspaceRoot ); + await output.drain(); process.exit(0); } @@ -100,6 +101,7 @@ export async function runOne( extraTargetDependencies, extraOptions ); + await output.drain(); process.exit(status); } } diff --git a/packages/nx/src/command-line/show/project.ts b/packages/nx/src/command-line/show/project.ts index 47ef0127c11..1f08301d95f 100644 --- a/packages/nx/src/command-line/show/project.ts +++ b/packages/nx/src/command-line/show/project.ts @@ -126,7 +126,5 @@ export async function showProjectHandler( } } - // TODO: Find a better fix for this - await new Promise((res) => setImmediate(res)); await output.drain(); } diff --git a/packages/nx/src/command-line/show/projects.ts b/packages/nx/src/command-line/show/projects.ts index 77f6112d18f..245dde83346 100644 --- a/packages/nx/src/command-line/show/projects.ts +++ b/packages/nx/src/command-line/show/projects.ts @@ -87,8 +87,6 @@ export async function showProjectsHandler( } } - // TODO: Find a better fix for this - await new Promise((res) => setImmediate(res)); await output.drain(); } diff --git a/packages/nx/src/tasks-runner/run-command.ts b/packages/nx/src/tasks-runner/run-command.ts index 2616015240f..5cff2cc5d0f 100644 --- a/packages/nx/src/tasks-runner/run-command.ts +++ b/packages/nx/src/tasks-runner/run-command.ts @@ -165,16 +165,27 @@ async function getTerminalOutputLifeCycle( console.log = createPatchedConsoleMethod(originalConsoleLog); console.error = createPatchedConsoleMethod(originalConsoleError); - const patchedWrite = (_chunk, _encoding, callback) => { + // Handle both overload signatures: dropping write(chunk, cb) would strand the + // callback and hang anything awaiting it, such as output.drain(). + // Typed rather than cast so the compiler keeps checking that. + const patchedWrite = ( + _chunk: string | Uint8Array, + encodingOrCallback?: BufferEncoding | ((err?: Error) => void), + callback?: (err?: Error) => void + ): boolean => { + const cb = + typeof encodingOrCallback === 'function' + ? encodingOrCallback + : callback; // Preserve original behavior around callback and return value, just in case - if (callback) { - callback(); + if (cb) { + cb(null); } return true; }; - process.stdout.write = patchedWrite as any; - process.stderr.write = patchedWrite as any; + process.stdout.write = patchedWrite; + process.stderr.write = patchedWrite; const { AppLifeCycle, restoreTerminal } = await handleImport( '../native/index.js', @@ -266,29 +277,36 @@ async function getTerminalOutputLifeCycle( /** * Patch stdout.write and stderr.write methods to pass Nx Cloud client logs to the TUI via the lifecycle */ - const createPatchedLogWrite = ( - originalWrite: - | typeof process.stdout.write - | typeof process.stderr.write, - isError: boolean - ): typeof process.stdout.write | typeof process.stderr.write => { - // @ts-ignore - return (chunk, encoding, callback) => { - if (isError) { - logDebug( - Buffer.isBuffer(chunk) - ? chunk.toString(encoding) - : chunk.toString() - ); - } else { - logDebug( - Buffer.isBuffer(chunk) - ? chunk.toString(encoding) - : chunk.toString() - ); - } + const createPatchedLogWrite = (): typeof process.stdout.write => { + // Handle both overload signatures: dropping write(chunk, cb) would strand the + // callback, and the callback would reach toString() below as the encoding. + // isEncoding also rejects the values toString() cannot take ('buffer', '', null). + return ( + chunk: string | Uint8Array, + encodingOrCallback?: BufferEncoding | ((err?: Error) => void), + callback?: (err?: Error) => void + ): boolean => { + const cb = + typeof encodingOrCallback === 'function' + ? encodingOrCallback + : callback; + const enc = + typeof encodingOrCallback === 'string' && + Buffer.isEncoding(encodingOrCallback) + ? encodingOrCallback + : undefined; + logDebug( + ArrayBuffer.isView(chunk) + ? Buffer.from( + chunk.buffer, + chunk.byteOffset, + chunk.byteLength + ).toString(enc) + : chunk.toString() + ); - // Check if the log came from the Nx Cloud client, otherwise invoke the original write method + // Nx Cloud logs are held back for the TUI; everything else is swallowed, + // since the TUI owns the terminal while it is running. const stackTrace = new Error().stack; const isNxCloudLog = stackTrace.includes(nxCloudClientDir); if (isNxCloudLog) { @@ -304,8 +322,8 @@ async function getTerminalOutputLifeCycle( } } // Preserve original behavior around callback and return value, just in case - if (callback) { - callback(); + if (cb) { + cb(null); } return true; }; @@ -325,8 +343,8 @@ async function getTerminalOutputLifeCycle( }; }; - process.stdout.write = createPatchedLogWrite(originalStdoutWrite, false); - process.stderr.write = createPatchedLogWrite(originalStderrWrite, true); + process.stdout.write = createPatchedLogWrite(); + process.stderr.write = createPatchedLogWrite(); // The cloud client calls console.log when NX_VERBOSE_LOGGING is set to true console.log = createPatchedConsoleMethod(originalConsoleLog); diff --git a/packages/nx/src/tasks-runner/task-orchestrator.spec.ts b/packages/nx/src/tasks-runner/task-orchestrator.spec.ts index f0edcc35d8e..9cef995d4cf 100644 --- a/packages/nx/src/tasks-runner/task-orchestrator.spec.ts +++ b/packages/nx/src/tasks-runner/task-orchestrator.spec.ts @@ -390,4 +390,45 @@ describe('TaskOrchestrator', () => { expect(orchestrator.cache.getBatch).toHaveBeenCalledTimes(1); }); }); + + describe('SIGINT output silencing', () => { + // The handler replaces process.stdout.write permanently and never restores it, + // so anything awaiting a write callback after Ctrl-C depends on this shape. + it('invokes the callback from either argument position', async () => { + const realStdoutWrite = process.stdout.write; + const realStderrWrite = process.stderr.write; + const realOn = process.on; + try { + const orchestrator: any = Object.create(TaskOrchestrator.prototype); + orchestrator.tuiEnabled = false; + orchestrator.runningContinuousTasks = new Map(); + orchestrator.cleanup = jest.fn(async () => {}); + orchestrator.resolveStopPromise = jest.fn(); + + const handlers: Record void> = {}; + jest.spyOn(process, 'on').mockImplementation((( + signal: string, + fn: any + ) => { + handlers[signal] = fn; + return process; + }) as any); + + orchestrator.setupSignalHandlers(); + handlers['SIGINT']('SIGINT'); + + const twoArg = jest.fn(); + expect((process.stdout.write as any)('x', twoArg)).toBe(true); + expect(twoArg).toHaveBeenCalled(); + + const threeArg = jest.fn(); + expect((process.stderr.write as any)('x', 'utf8', threeArg)).toBe(true); + expect(threeArg).toHaveBeenCalled(); + } finally { + process.stdout.write = realStdoutWrite; + process.stderr.write = realStderrWrite; + process.on = realOn; + } + }); + }); }); diff --git a/packages/nx/src/tasks-runner/task-orchestrator.ts b/packages/nx/src/tasks-runner/task-orchestrator.ts index 3495b2abe25..187b3702b57 100644 --- a/packages/nx/src/tasks-runner/task-orchestrator.ts +++ b/packages/nx/src/tasks-runner/task-orchestrator.ts @@ -1895,12 +1895,23 @@ export class TaskOrchestrator { // Silence output — pnpm (and similar wrappers) may exit before nx // finishes cleanup, returning the shell prompt. Any output after // that point would appear after the prompt. - const noop = (_chunk, _encoding, callback) => { - if (callback) callback(); + // Handle both overload signatures: dropping write(chunk, cb) would strand + // the callback and hang anything awaiting it, such as output.drain(). + // Typed rather than cast so the compiler keeps checking that. + const noop = ( + _chunk: string | Uint8Array, + encodingOrCallback?: BufferEncoding | ((err?: Error) => void), + callback?: (err?: Error) => void + ): boolean => { + const cb = + typeof encodingOrCallback === 'function' + ? encodingOrCallback + : callback; + if (cb) cb(null); return true; }; - process.stdout.write = noop as any; - process.stderr.write = noop as any; + process.stdout.write = noop; + process.stderr.write = noop; } this.cleanup().finally(() => { if (this.resolveStopPromise) { diff --git a/packages/nx/src/utils/output.spec.ts b/packages/nx/src/utils/output.spec.ts new file mode 100644 index 00000000000..5d7259bc620 --- /dev/null +++ b/packages/nx/src/utils/output.spec.ts @@ -0,0 +1,196 @@ +import { Writable } from 'stream'; +import { output } from './output'; + +/** + * A stdout stand-in that never completes a write until the test releases or fails + * it, so queued bytes stay queued — the state a slow pipe reader produces. + */ +function stalledStdout(highWaterMark: number) { + const pending: Array<(err?: Error) => void> = []; + let received = ''; + const stream = new Writable({ + highWaterMark, + write(chunk, _enc, cb) { + pending.push((err) => { + if (!err) received += chunk.toString(); + cb(err); + }); + }, + }); + return { + stream, + get received() { + return received; + }, + release: () => { + while (pending.length) pending.shift()!(); + }, + // Node hands EPIPE to the write callback first, then emits 'error'. + failPending: (err: Error) => { + while (pending.length) pending.shift()!(err); + }, + }; +} + +describe('output.drain', () => { + const realStdout = process.stdout; + + function useStdout(stream: Writable) { + Object.defineProperty(process, 'stdout', { + value: stream, + configurable: true, + }); + } + + afterEach(() => { + Object.defineProperty(process, 'stdout', { + value: realStdout, + configurable: true, + }); + }); + + it('resolves immediately when nothing is queued', async () => { + const { stream } = stalledStdout(1000); + useStdout(stream); + + await expect(output.drain()).resolves.toBeUndefined(); + }); + + // Regression: `writableNeedDrain` is only set past the high-water mark, so a + // queue shorter than it used to resolve drain() instantly and let process.exit() + // discard the bytes. 50 bytes against a 1000-byte mark reproduces that exactly. + it('waits for a queue shorter than the high-water mark', async () => { + const stalled = stalledStdout(1000); + useStdout(stalled.stream); + + stalled.stream.write('a'.repeat(50)); + expect(stalled.stream.writableNeedDrain).toBe(false); + expect(stalled.stream.writableLength).toBeGreaterThan(0); + + let drained = false; + const promise = output.drain().then(() => (drained = true)); + + await new Promise((res) => setImmediate(res)); + expect(drained).toBe(false); + + stalled.release(); + await promise; + expect(drained).toBe(true); + expect(stalled.received).toBe('a'.repeat(50)); + }); + + it('waits for a queue longer than the high-water mark', async () => { + const stalled = stalledStdout(100); + useStdout(stalled.stream); + + stalled.stream.write('b'.repeat(500)); + expect(stalled.stream.writableNeedDrain).toBe(true); + + let drained = false; + const promise = output.drain().then(() => (drained = true)); + + await new Promise((res) => setImmediate(res)); + expect(drained).toBe(false); + + stalled.release(); + await promise; + expect(drained).toBe(true); + expect(stalled.received).toBe('b'.repeat(500)); + }); + + // `nx ... | head` leaves the write end open with no reader; the EPIPE must not + // hang the drain or surface as an unhandled 'error' event. + it('resolves instead of hanging when the stream errors', async () => { + const stalled = stalledStdout(1000); + useStdout(stalled.stream); + + stalled.stream.write('c'.repeat(50)); + const promise = output.drain(); + + // Node's real ordering: write callback first, then the 'error' event. Emitting + // 'error' directly, or dropping the listener assertion, each let a + // listener-removing refactor pass here while crashing on an actual pipe. + stalled.failPending(Object.assign(new Error('EPIPE'), { code: 'EPIPE' })); + expect(stalled.stream.listenerCount('error')).toBe(1); + await expect(promise).resolves.toBeUndefined(); + // Settle the deferred detach here so a too-early removal surfaces in this test + // rather than as an unhandled error attributed to whichever test runs next. + await new Promise((res) => setImmediate(res)); + }); + + // process.nextTick drains before the 'error' emit, so a cleanup deferred that far + // detaches too early and the EPIPE goes unhandled. Only a macrotask is late enough. + it('keeps the listener attached through the nextTick queue', async () => { + const stalled = stalledStdout(1000); + useStdout(stalled.stream); + + stalled.stream.write('f'.repeat(50)); + const promise = output.drain(); + stalled.release(); + + await new Promise((res) => process.nextTick(res)); + expect(stalled.stream.listenerCount('error')).toBe(1); + + await promise; + await new Promise((res) => setImmediate(res)); + expect(stalled.stream.listenerCount('error')).toBe(0); + }); + + // The cleanup is deferred, so it must target the stream drain() attached to + // rather than re-reading process.stdout after a caller has swapped it. + it('detaches from the stream it attached to, not the current process.stdout', async () => { + const stalled = stalledStdout(1000); + useStdout(stalled.stream); + + stalled.stream.write('e'.repeat(50)); + const promise = output.drain(); + // Guards against the assertion below passing vacuously on an early return. + expect(stalled.stream.listenerCount('error')).toBe(1); + stalled.release(); + await promise; + + useStdout(realStdout); + await new Promise((res) => setImmediate(res)); + + expect(stalled.stream.listenerCount('error')).toBe(0); + }); + + // A positional (chunk, encoding, callback) stdout patch that does not normalize — + // as any third-party wrapper may be — drops a two-argument write's callback. + it('resolves when process.stdout.write is patched positionally', async () => { + const stalled = stalledStdout(1000); + useStdout(stalled.stream); + + stalled.stream.write('g'.repeat(50)); + let writes = 0; + (stalled.stream as any).write = ( + _chunk: unknown, + _encoding: unknown, + callback?: () => void + ) => { + writes++; + if (callback) callback(); + return true; + }; + + await expect(output.drain()).resolves.toBeUndefined(); + // Also guards against an early return: drain must have reached the patched + // writer rather than resolving before it queued anything. + expect(writes).toBe(1); + }, 5000); + + it('leaves no error listener behind', async () => { + const stalled = stalledStdout(1000); + useStdout(stalled.stream); + + for (let i = 0; i < 3; i++) { + stalled.stream.write('d'.repeat(50)); + const promise = output.drain(); + stalled.release(); + await promise; + } + await new Promise((res) => setImmediate(res)); + + expect(stalled.stream.listenerCount('error')).toBe(0); + }); +}); diff --git a/packages/nx/src/utils/output.ts b/packages/nx/src/utils/output.ts index d4334f910ba..c58359f8722 100644 --- a/packages/nx/src/utils/output.ts +++ b/packages/nx/src/utils/output.ts @@ -357,13 +357,35 @@ class CLIOutput { this.addNewline(); } + /** + * Waits for queued `process.stdout` to reach the fd. Does not cover stderr, and + * resolves without flushing if something has replaced `process.stdout.write` — + * the SIGINT silencer in task-orchestrator.ts does exactly that, on purpose. + */ drain(): Promise { return new Promise((resolve) => { - if (process.stdout.writableNeedDrain) { - process.stdout.once('drain', resolve); - } else { + // Captured once: the deferred cleanup below must detach from the stream it + // attached to, not from whatever `process.stdout` is a macrotask later. + const stream = process.stdout; + // `writableNeedDrain` is only set once the queue passes the high-water mark, + // so a shorter queue needs the write callback instead. Waiting on the 'drain' + // event alone lets `process.exit()` discard up to highWaterMark of output. + if (stream.writableLength === 0) { resolve(); + return; } + // The reader may already be gone (`nx ... | head`); an unhandled EPIPE would + // kill the run. Detach via setImmediate — not in the write callback and not via + // process.nextTick: both run before the 'error' event, so the crash comes back. + const onError = () => resolve(); + stream.on('error', onError); + // Encoding passed explicitly: a positional (chunk, encoding, callback) stdout + // patch that does not normalize drops a two-argument write's callback, and this + // would never resolve. The patches in this repo normalize; third-party ones may not. + stream.write('', 'utf8', () => { + resolve(); + setImmediate(() => stream.removeListener('error', onError)); + }); }); } }