diff --git a/.bumpy/run-piped-output-drain.md b/.bumpy/run-piped-output-drain.md new file mode 100644 index 000000000..0c2576c4f --- /dev/null +++ b/.bumpy/run-piped-output-drain.md @@ -0,0 +1,5 @@ +--- +varlock: patch +--- + +Fix `varlock run` and `varlock proxy run` dropping piped output: output still unread when the command exits, output from background processes it started, and output that a slow reader had not consumed within 1 second diff --git a/packages/varlock-website/src/content/docs/reference/cli/load-and-run.mdx b/packages/varlock-website/src/content/docs/reference/cli/load-and-run.mdx index db60d4504..e5e95720d 100644 --- a/packages/varlock-website/src/content/docs/reference/cli/load-and-run.mdx +++ b/packages/varlock-website/src/content/docs/reference/cli/load-and-run.mdx @@ -146,6 +146,8 @@ varlock run -- sh -c 'echo $MY_VAR' # ✅ will work Detection is per stream. For example `varlock run -- node app.js | tee log.txt` redacts stdout (piped) while stderr (still attached to your terminal) passes through. +When output is piped, varlock reads everything the command writes before exiting, including output from background processes it started that still hold the pipe. To avoid hanging on a long-lived background process (for example `sh -c 'server &'`), it waits at most 2 seconds after the command exits, then closes the pipe. Anything written after that is dropped, and the background process gets `SIGPIPE` on its next write. + To override the auto-detection: ```bash # force-disable redaction even when output is piped/redirected diff --git a/packages/varlock/src/cli/commands/proxy.command.ts b/packages/varlock/src/cli/commands/proxy.command.ts index ec6cbb408..386f46437 100644 --- a/packages/varlock/src/cli/commands/proxy.command.ts +++ b/packages/varlock/src/cli/commands/proxy.command.ts @@ -90,7 +90,9 @@ import type { ProxyManagedItem, ProxyRule, ProxyTransformSchemeDef } from '../.. import { generateProxyPlaceholderForItem } from '../../proxy/placeholder'; import { isVarlockReservedKey } from '../../env-graph/lib/reserved-vars'; import { resetRedactionMap } from '../../runtime/env'; -import { resolveStdoutRedaction, pipeRedactedStreams, setParentRedactedStreamsEnv } from '../helpers/stdout-redaction'; +import { + resolveStdoutRedaction, pipeRedactedStreams, setParentRedactedStreamsEnv, flushStdio, +} from '../helpers/stdout-redaction'; import { type TypedGunshiCommandFn } from '../helpers/gunshi-type-utils'; import { CliExitError } from '../helpers/exit-error'; import { reportChildCommandError } from '../helpers/child-exit'; @@ -1678,6 +1680,10 @@ async function awaitProxiedChild( opts.onSummary?.(); } + // cleanup (audit log flush, session record) is done: a signal held during it, or one + // arriving while we flush output to a slow reader, now ends varlock + opts.signalForwarder.beginFinalFlush(); + await flushStdio(); return exitCode; } diff --git a/packages/varlock/src/cli/commands/run.command.ts b/packages/varlock/src/cli/commands/run.command.ts index d324d9764..a5641c580 100644 --- a/packages/varlock/src/cli/commands/run.command.ts +++ b/packages/varlock/src/cli/commands/run.command.ts @@ -9,7 +9,9 @@ import { } from '../helpers/error-checks'; import { getCliItemFilter } from '../helpers/item-filter'; import { type TypedGunshiCommandFn } from '../helpers/gunshi-type-utils'; -import { resolveStdoutRedaction, pipeRedactedStreams, setParentRedactedStreamsEnv } from '../helpers/stdout-redaction'; +import { + resolveStdoutRedaction, pipeRedactedStreams, setParentRedactedStreamsEnv, flushStdio, +} from '../helpers/stdout-redaction'; import { flushSchemaLoadedEvent } from '../helpers/telemetry'; import { buildInjectedBlobEnv } from '../helpers/injected-env-blob'; import { resolveInjectMode } from '../helpers/inject-mode'; @@ -330,6 +332,9 @@ export const commandFn: TypedGunshiCommandFn = async (ctx) = } if (!isWatchEnabled) { + // no cleanup step here: a signal from now on ends varlock, even mid-flush + signalForwarder.beginFinalFlush(); + await flushStdio(); return gracefulExit(exitCode); } else { console.error('... watching for changes ...'); diff --git a/packages/varlock/src/cli/helpers/stdout-redaction.ts b/packages/varlock/src/cli/helpers/stdout-redaction.ts index 0b6e9784b..be7acf517 100644 --- a/packages/varlock/src/cli/helpers/stdout-redaction.ts +++ b/packages/varlock/src/cli/helpers/stdout-redaction.ts @@ -47,6 +47,28 @@ export function resolveStdoutRedaction(opts: { }; } +/** + * Streams with a `write()` that returned false and no `drain` since. We track this ourselves + * because Bun's stdout/stderr hand writes straight to a native sink: `writableLength` and + * `writableNeedDrain` stay 0/false and an empty write's callback fires immediately, even with + * megabytes still queued. A `false` return followed by `drain` is the only signal that works + * in both Node and Bun. + */ +const streamsAwaitingDrain = new Set(); +const drainListeners = new WeakSet(); + +function drainTrackedWriter(stream: NodeJS.WritableStream) { + if (!drainListeners.has(stream)) { + drainListeners.add(stream); + stream.on('drain', () => streamsAwaitingDrain.delete(stream)); + } + return { + write(text: string) { + if (!stream.write(text)) streamsAwaitingDrain.add(stream); + }, + }; +} + /** * Wire the shared chunk-boundary-buffered redactor onto whichever of the child's streams the * plan marks for redaction (so a secret split across two chunks is still caught). No-ops for @@ -57,12 +79,12 @@ export function pipeRedactedStreams( plan: StdoutRedactionPlan, ): void { if (plan.redactStdout && commandProcess.stdout) { - const writer = createRedactedStreamWriter(process.stdout); + const writer = createRedactedStreamWriter(drainTrackedWriter(process.stdout)); commandProcess.stdout.on('data', writer.write); commandProcess.stdout.on('close', writer.flush); } if (plan.redactStderr && commandProcess.stderr) { - const writer = createRedactedStreamWriter(process.stderr); + const writer = createRedactedStreamWriter(drainTrackedWriter(process.stderr)); commandProcess.stderr.on('data', writer.write); commandProcess.stderr.on('close', writer.flush); } @@ -80,3 +102,29 @@ export function setParentRedactedStreamsEnv(env: Record { + await Promise.all([process.stdout, process.stderr].map((stream) => new Promise((resolve) => { + if (!stream.writable || stream.destroyed) { + resolve(); + return; + } + const done = () => { + stream.off('error', done); + resolve(); + }; + // then an empty write as a barrier for anything written since (Node only; Bun's fires at once) + const afterDrain = () => stream.write('', done); + stream.once('error', done); + if (streamsAwaitingDrain.has(stream)) stream.once('drain', afterDrain); + else afterDrain(); + }))); +} diff --git a/packages/varlock/src/lib/child-signals.ts b/packages/varlock/src/lib/child-signals.ts index c8df26157..5f731a9c2 100644 --- a/packages/varlock/src/lib/child-signals.ts +++ b/packages/varlock/src/lib/child-signals.ts @@ -1,4 +1,6 @@ import { openSync, closeSync } from 'node:fs'; +import { constants as osConstants } from 'node:os'; +import { gracefulExit } from 'exit-hook'; /** * Signals we forward to a long-running child so it can shut down gracefully: the @@ -84,8 +86,19 @@ export interface ChildSignalForwarder { * attach is forwarded now, so an `await` in that window can't drop it. Idempotent. */ attach(child: ChildLike): void; - /** Call once the child has exited and been reaped, so we never signal a recycled pid. */ + /** + * Call once the child has exited and been reaped, so we never signal a recycled pid. + * A signal arriving after this is held (not forwarded, not acted on) so the caller's + * cleanup can finish; `beginFinalFlush()` then acts on it. + */ detach(): void; + /** + * Call after post-child cleanup, right before the final (unbounded) stdout/stderr flush. + * If a signal arrived since `detach()`, varlock exits now with 128+N; any later signal + * exits immediately. Our claimed listeners replaced the default "terminate" action, so + * without this a stalled output reader would make varlock unkillable. + */ + beginFinalFlush(): void; /** * Whether varlock received (and forwarded) a shutdown signal (SIGINT/SIGTERM/SIGQUIT, * not SIGHUP) while the child ran. A child that handles e.g. SIGINT and then exits with @@ -131,6 +144,12 @@ export function createChildSignalForwarder(): ChildSignalForwarder { // signals received before attach(); replayed once we know the child const pendingSignals: Array = []; + let finalFlushStarted = false; + // first signal received after the child exited, before beginFinalFlush() + let heldSignal: NodeJS.Signals | undefined; + // exit as if the signal had killed us (exit-hook ignores repeat calls) + const exitForSignal = (signal: NodeJS.Signals) => gracefulExit(128 + (osConstants.signals[signal] ?? 0)); + const signalChild = (signal: NodeJS.Signals | number) => { if (childExited || !child?.pid) return; try { @@ -150,6 +169,12 @@ export function createChildSignalForwarder(): ChildSignalForwarder { FORWARDED_SIGNALS.forEach((signal) => { const forwardSignal = () => { + // the child is gone: nothing to forward to (see detach / beginFinalFlush) + if (childExited) { + if (finalFlushStarted) exitForSignal(signal); + else heldSignal ??= signal; + return; + } if (SHUTDOWN_SIGNALS.includes(signal)) receivedShutdownSignal = true; if (!child) { pendingSignals.push(signal); @@ -186,6 +211,10 @@ export function createChildSignalForwarder(): ChildSignalForwarder { childExited = true; if (forceKillTimer) clearTimeout(forceKillTimer); }, + beginFinalFlush() { + finalFlushStarted = true; + if (heldSignal) exitForSignal(heldSignal); + }, get receivedShutdownSignal() { return receivedShutdownSignal; }, diff --git a/packages/varlock/src/lib/exec.ts b/packages/varlock/src/lib/exec.ts index 038a8cbc2..97f5992f0 100644 --- a/packages/varlock/src/lib/exec.ts +++ b/packages/varlock/src/lib/exec.ts @@ -20,6 +20,37 @@ interface ExecOptions { * grandchildren are terminated too. Ignored on Windows. */ detached?: boolean; + /** + * After the child exits, how long to keep waiting for its piped stdout/stderr to close + * before giving up and destroying them. A descendant that inherited the pipe (e.g. a + * backgrounded server) can hold it open indefinitely, so this must be bounded. + */ + outputDrainTimeoutMs?: number; +} + +const DEFAULT_OUTPUT_DRAIN_TIMEOUT_MS = 2000; + +/** + * Wait for the given streams to close, destroying any still open after `timeoutMs` + * (which emits `close`, so listeners such as the redaction writer still flush). + */ +function waitForPipesToClose(streams: Array, timeoutMs: number): Promise { + const open = streams.filter((s): s is Readable => !!s && !s.closed); + if (!open.length) return Promise.resolve(); + return new Promise((resolve) => { + const timer = setTimeout(() => { + for (const s of open) { + if (!s.closed) s.destroy(); + } + }, timeoutMs); + Promise.all(open.map((s) => new Promise((r) => { + if (s.closed) r(); + else s.once('close', () => r()); + }))).then(() => { + clearTimeout(timer); + resolve(); + }); + }); } /** @@ -296,7 +327,7 @@ export function exec( ); }); - childProcess.on('exit', (code, signal) => { + const settle = (code: number | null, signal: NodeJS.Signals | null) => { // Windows special case: exit code 1 without error event might be ENOENT if (process.platform === 'win32' && code === 1 && !errorEmitted && !existsSync(resolvedCommand)) { const error: any = new Error(`Command not found: ${command}`); @@ -324,6 +355,15 @@ export function exec( } else { resolve(exitResult); } + }; + + childProcess.on('exit', (code, signal) => { + // `exit` can fire before the child's piped output has been read, so settle only once + // those pipes close (or the post-exit grace period runs out). + waitForPipesToClose( + [childProcess.stdout, childProcess.stderr], + options.outputDrainTimeoutMs ?? DEFAULT_OUTPUT_DRAIN_TIMEOUT_MS, + ).then(() => settle(code, signal)); }); }) as Promise & Partial & { kill: (signal?: number | NodeJS.Signals) => boolean }; diff --git a/packages/varlock/src/lib/test/child-signals.test.ts b/packages/varlock/src/lib/test/child-signals.test.ts index 47547c111..ef43d42d1 100644 --- a/packages/varlock/src/lib/test/child-signals.test.ts +++ b/packages/varlock/src/lib/test/child-signals.test.ts @@ -1,8 +1,11 @@ import { describe, test, expect, afterEach, vi, } from 'vitest'; +import { gracefulExit } from 'exit-hook'; import { claimSignals, createChildSignalForwarder, FORWARDED_SIGNALS } from '../child-signals'; +vi.mock('exit-hook', () => ({ gracefulExit: vi.fn() })); + const SIGNALS: Array = ['SIGTERM', 'SIGINT']; describe('claimSignals', () => { @@ -52,6 +55,49 @@ describe('claimSignals', () => { } }); + describe('after the child exits', () => { + afterEach(() => vi.mocked(gracefulExit).mockClear()); + + function exitedForwarder() { + const killSpy = vi.spyOn(process, 'kill').mockImplementation(() => true); + const child = { pid: 424242, kill: vi.fn((_signal?: number | NodeJS.Signals) => true) }; + const forwarder = createChildSignalForwarder(); + forwarder.attach(child); + forwarder.detach(); + return { forwarder, child, killSpy }; + } + + test('a signal during cleanup is held, then acted on by beginFinalFlush()', () => { + const { forwarder, child, killSpy } = exitedForwarder(); + try { + process.emit('SIGTERM', 'SIGTERM'); + process.emit('SIGINT', 'SIGINT'); + expect(gracefulExit).not.toHaveBeenCalled(); + expect(child.kill).not.toHaveBeenCalled(); + expect(killSpy).not.toHaveBeenCalled(); + + forwarder.beginFinalFlush(); + // the first held signal decides the exit status + expect(gracefulExit).toHaveBeenCalledTimes(1); + expect(gracefulExit).toHaveBeenCalledWith(143); + } finally { + killSpy.mockRestore(); + } + }); + + test('a signal during the final flush exits immediately with 128+N', () => { + const { forwarder, killSpy } = exitedForwarder(); + try { + forwarder.beginFinalFlush(); + expect(gracefulExit).not.toHaveBeenCalled(); + process.emit('SIGINT', 'SIGINT'); + expect(gracefulExit).toHaveBeenCalledWith(130); + } finally { + killSpy.mockRestore(); + } + }); + }); + test('createChildSignalForwarder leaves exactly one listener per forwarded signal', () => { // mimic exit-hook, which registers its own SIGINT/SIGTERM listeners at module load const exitHookLike = () => undefined; diff --git a/packages/varlock/src/lib/test/exec.test.ts b/packages/varlock/src/lib/test/exec.test.ts index 6a159d016..6fee38fbd 100644 --- a/packages/varlock/src/lib/test/exec.test.ts +++ b/packages/varlock/src/lib/test/exec.test.ts @@ -69,6 +69,32 @@ describe('exec', () => { expect(output.trim()).toBe('err-output'); }); + it('should wait for a descendant still writing to the inherited pipe', async () => { + // the child exits immediately, but a grandchild holding its stdout writes 300ms later + const script = ` + require('node:child_process').spawn(process.execPath, ['-e', 'setTimeout(() => console.log("late"), 300)'], { stdio: ['ignore', 'inherit', 'inherit'] }).unref(); + console.log('early'); + `; + const childProcess = exec('node', ['-e', script], { stdout: 'pipe', stderr: 'pipe' }); + const chunks: Array = []; + childProcess.stdout!.on('data', (d) => chunks.push(d.toString())); + + await childProcess; + expect(chunks.join('').split('\n').filter(Boolean)).toEqual(['early', 'late']); + }); + + it('should stop waiting for a descendant holding the pipe after the drain timeout', async () => { + const script = ` + require('node:child_process').spawn(process.execPath, ['-e', 'setTimeout(() => {}, 10000)'], { stdio: ['ignore', 'inherit', 'inherit'] }).unref(); + `; + const startMs = Date.now(); + const childProcess = exec('node', ['-e', script], { stdout: 'pipe', stderr: 'pipe', outputDrainTimeoutMs: 300 }); + const result = await childProcess; + expect(result.exitCode).toBe(0); + expect(Date.now() - startMs).toBeLessThan(5000); + expect(childProcess.stdout!.destroyed).toBe(true); + }); + it('should resolve with inherited stdio', async () => { const result = await exec('node', ['-e', 'process.exit(0)'], { stdio: 'inherit', diff --git a/smoke-tests/tests/run-output-drain.test.ts b/smoke-tests/tests/run-output-drain.test.ts new file mode 100644 index 000000000..639ed9bf2 --- /dev/null +++ b/smoke-tests/tests/run-output-drain.test.ts @@ -0,0 +1,94 @@ +import { describe, test, expect } from 'vitest'; +import { execSync, spawn } from 'node:child_process'; +import { join } from 'node:path'; +import { VARLOCK_CLI } from '../helpers/run-varlock.js'; + +// `varlock run` with piped (redacted) output must not exit before the child's output has +// been read and written onward (#1279). + +const BASIC_CWD = join(import.meta.dirname, '..', 'smoke-test-basic'); +const BIG = 2 * 1024 * 1024; +const WRITE_BIG = `process.stdout.write(Buffer.alloc(${BIG}, 120), () => process.stderr.write('child-done\\n'))`; + +function hasBun(): boolean { + try { + execSync('bun --version', { stdio: 'ignore' }); + return true; + } catch { + return false; + } +} + +function spawnVarlock(runtime: 'node' | 'bun', command: Array) { + return spawn(runtime === 'node' ? process.execPath : 'bun', [VARLOCK_CLI, 'run', '--', ...command], { + cwd: BASIC_CWD, + stdio: ['ignore', 'pipe', 'pipe'], + }); +} + +function waitForExit(child: ReturnType, timeoutMs = 15_000) { + return new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => { + const timer = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error('varlock did not exit')); + }, timeoutMs); + child.on('exit', (code, signal) => { + clearTimeout(timer); + resolve({ code, signal }); + }); + }); +} + +describe('varlock run piped output draining', () => { + test('keeps output a background process writes after the child exits', async () => { + const script = ` + require('node:child_process').spawn(process.execPath, ['-e', 'setTimeout(() => console.log("late line"), 300)'], { stdio: ['ignore', 'inherit', 'inherit'] }).unref(); + console.log('early line'); + `; + const child = spawnVarlock('node', ['node', '-e', script]); + let stdout = ''; + child.stdout!.on('data', (d) => { + stdout += d.toString(); + }); + const { code } = await waitForExit(child); + expect(code).toBe(0); + expect(stdout.split('\n').filter(Boolean)).toEqual(['early line', 'late line']); + }); + + for (const runtime of ['node', 'bun'] as const) { + test.skipIf(runtime === 'bun' && !hasBun())(`delivers all output to a slow reader under ${runtime}`, async () => { + const child = spawnVarlock(runtime, ['node', '-e', WRITE_BIG]); + const exited = waitForExit(child); + // don't read for longer than exit-hook's 1s stdio flush window + await new Promise((r) => { + setTimeout(r, 2500); + }); + let bytes = 0; + child.stdout!.on('data', (d: Buffer) => { + bytes += d.length; + }); + const { code } = await exited; + expect(code).toBe(0); + expect(bytes).toBe(BIG); + }); + } + + test('SIGTERM still exits varlock while it waits on a stalled reader', async () => { + const child = spawnVarlock('node', ['node', '-e', WRITE_BIG]); + const exited = waitForExit(child); + // stdout is never read; wait until the child has handed off all its output and exited + await new Promise((resolve) => { + let stderr = ''; + child.stderr!.on('data', (d) => { + stderr += d.toString(); + if (stderr.includes('child-done')) resolve(); + }); + }); + await new Promise((r) => { + setTimeout(r, 500); + }); + child.kill('SIGTERM'); + const { code } = await exited; + expect(code).toBe(143); + }); +});