diff --git a/.bumpy/run-pipe-drain.md b/.bumpy/run-pipe-drain.md new file mode 100644 index 000000000..a8b22776d --- /dev/null +++ b/.bumpy/run-pipe-drain.md @@ -0,0 +1,5 @@ +--- +"varlock": patch +--- + +Preserve redacted child output after process exit, with a bounded wait for inherited pipes to close. 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..3fc33dfec 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 @@ -180,6 +180,8 @@ Note: When the `__VARLOCK_ENV` blob is omitted, framework integrations that rely This makes `varlock run` safe to use as a container `ENTRYPOINT` (including PID 1), so `docker stop` / pod termination reaches your app gracefully instead of waiting out the grace period and being `SIGKILL`ed. +After the child exits, `varlock run` and `varlock proxy run` allow up to two seconds for piped output to finish draining through redaction. If a descendant keeps a pipe open or the output destination remains blocked, varlock closes its pipe readers and reports that output may be incomplete. An otherwise successful command exits non-zero; an existing non-zero status or a forwarded shutdown keeps the child's exit status. This limit does not shorten a running child's lifetime or disable redaction. + One PID 1 duty varlock does not take on is reaping orphaned processes. If your app spawns subprocesses that can outlive their parent, they are reparented to PID 1 and stay as zombies after they exit. In that case run a minimal init above varlock, for example `ENTRYPOINT ["tini", "--", "varlock", "run", "--"]` or `docker run --init`. Signal forwarding works the same either way. See the [Docker guide](/integrations/docker/#pid-1-signals-and-zombie-processes). By default varlock forwards and then waits indefinitely for the child. It does **not** impose its own kill deadline, leaving that decision to the orchestrator or operator (a forwarded signal isn't always terminal, since many programs use `SIGHUP` to reload). If you want varlock to escalate to `SIGKILL` when the child hasn't exited within a grace period, set the `_VARLOCK_FORCE_KILL_TIMEOUT_MS` environment variable to a number of milliseconds. diff --git a/packages/varlock/src/cli/commands/proxy.command.ts b/packages/varlock/src/cli/commands/proxy.command.ts index ec6cbb408..2a4b6ce96 100644 --- a/packages/varlock/src/cli/commands/proxy.command.ts +++ b/packages/varlock/src/cli/commands/proxy.command.ts @@ -91,6 +91,7 @@ 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 { drainChildOutput } from '../helpers/drain-child-output'; import { type TypedGunshiCommandFn } from '../helpers/gunshi-type-utils'; import { CliExitError } from '../helpers/exit-error'; import { reportChildCommandError } from '../helpers/child-exit'; @@ -1666,6 +1667,7 @@ async function awaitProxiedChild( } finally { // child has exited and been reaped: stop forwarding (avoid signaling a recycled pid) opts.signalForwarder.detach(); + const shutdownRequested = opts.signalForwarder.receivedShutdownSignal; // Tear down the container sandbox before other cleanup, so nothing lingers. if (opts.sandboxTeardown) { try { @@ -1676,6 +1678,7 @@ async function awaitProxiedChild( } await opts.cleanup(); opts.onSummary?.(); + await drainChildOutput(commandProcess, { exitCode, shutdownRequested }); } return exitCode; diff --git a/packages/varlock/src/cli/commands/run.command.ts b/packages/varlock/src/cli/commands/run.command.ts index d324d9764..4ebad44f1 100644 --- a/packages/varlock/src/cli/commands/run.command.ts +++ b/packages/varlock/src/cli/commands/run.command.ts @@ -15,6 +15,7 @@ import { buildInjectedBlobEnv } from '../helpers/injected-env-blob'; import { resolveInjectMode } from '../helpers/inject-mode'; import { CliExitError } from '../helpers/exit-error'; import { reportChildCommandError } from '../helpers/child-exit'; +import { drainChildOutput } from '../helpers/drain-child-output'; import { evaluateInjectedEnvReuse, getUseInjectedEnvMode, USE_INJECTED_ENV_VAR } from '../../lib/injected-env-reuse'; import { getFrozenEnvFilePath, USE_FROZEN_ENV_VAR } from '../../lib/frozen-env-file'; import { applyFrozenArg, frozenEnvErrorToCliExitError } from '../helpers/frozen-env-cli'; @@ -317,6 +318,8 @@ export const commandFn: TypedGunshiCommandFn = async (ctx) = } finally { // child has exited and been reaped: stop forwarding (avoid signaling a recycled pid) signalForwarder.detach(); + const shutdownRequested = signalForwarder.receivedShutdownSignal; + await drainChildOutput(commandProcess, { exitCode, shutdownRequested }); } if (isWatchEnabled) { diff --git a/packages/varlock/src/cli/helpers/drain-child-output.ts b/packages/varlock/src/cli/helpers/drain-child-output.ts new file mode 100644 index 000000000..fc5ec0f5d --- /dev/null +++ b/packages/varlock/src/cli/helpers/drain-child-output.ts @@ -0,0 +1,76 @@ +import type { Readable, Writable } from 'node:stream'; +import { CliExitError } from './exit-error'; + +const drainTimeoutMs = 2000; +const outputDrains = new WeakMap Promise>(); + +export function trackChildOutput(source: object, destination: Writable) { + const pending = new Set>(); + let writeError: unknown; + outputDrains.set(source, async () => { + await Promise.all(pending); + if (writeError) throw writeError; + }); + return { + write(text: string) { + const written = new Promise((resolve, reject) => { + destination.write(text, (error) => { + if (error) reject(error); + else resolve(); + }); + }).catch((error: NodeJS.ErrnoException) => { + if (destination === process.stderr && (error.code === 'EPIPE' || (destination.errored as NodeJS.ErrnoException | null)?.code === 'EPIPE')) return; + writeError ??= error; + }); + pending.add(written); + written.then(() => pending.delete(written)); + }, + }; +} + +export async function drainChildOutput(commandProcess: { + stdout?: Readable; + stderr?: Readable; +}, completion: { exitCode?: number; shutdownRequested?: boolean } = {}): Promise { + const streams = [commandProcess.stdout, commandProcess.stderr] + .filter((stream): stream is Readable => stream !== undefined); + if (!streams.length) return; + + const cleanup: Array<() => void> = []; + const closed = streams.map((stream) => new Promise((resolve) => { + if (stream.closed) { + resolve(); + } else { + stream.once('close', resolve); + cleanup.push(() => { + stream.off('close', resolve); + }); + } + })); + let timer: ReturnType | undefined; + const deadline = new Promise((_resolve, reject) => { + timer = setTimeout(() => { + for (const stream of streams) stream.destroy(); + reject(new CliExitError(`Child output did not finish draining within ${drainTimeoutMs}ms`, { + details: 'The child has exited, but an inherited pipe or output destination is still open. Output may be incomplete.', + forceExit: true, + })); + }, drainTimeoutMs); + }); + + try { + await Promise.race([ + (async () => { + await Promise.all(closed); + await Promise.all(streams.map((stream) => outputDrains.get(stream)?.())); + })(), + deadline, + ]); + } catch (error) { + if (!(error instanceof CliExitError) || (!completion.exitCode && !completion.shutdownRequested)) throw error; + console.error(error.getFormattedOutput()); + } finally { + clearTimeout(timer); + for (const remove of cleanup) remove(); + } +} diff --git a/packages/varlock/src/cli/helpers/stdout-redaction.ts b/packages/varlock/src/cli/helpers/stdout-redaction.ts index 0b6e9784b..0350cb83b 100644 --- a/packages/varlock/src/cli/helpers/stdout-redaction.ts +++ b/packages/varlock/src/cli/helpers/stdout-redaction.ts @@ -1,6 +1,7 @@ import { CliExitError } from './exit-error'; import { createRedactedStreamWriter, PARENT_REDACTED_STREAMS_ENV_VAR } from '../../runtime/lib/redact-stream'; import { parseEnvToggle } from '../../runtime/lib/env-toggle'; +import { trackChildOutput } from './drain-child-output'; export { REDACT_STDOUT_ARG } from './redact-stdout-arg'; export { parseEnvToggle }; @@ -57,12 +58,12 @@ export function pipeRedactedStreams( plan: StdoutRedactionPlan, ): void { if (plan.redactStdout && commandProcess.stdout) { - const writer = createRedactedStreamWriter(process.stdout); + const writer = createRedactedStreamWriter(trackChildOutput(commandProcess.stdout, 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(trackChildOutput(commandProcess.stderr, process.stderr)); commandProcess.stderr.on('data', writer.write); commandProcess.stderr.on('close', writer.flush); } diff --git a/packages/varlock/src/cli/helpers/test/drain-child-output.test.ts b/packages/varlock/src/cli/helpers/test/drain-child-output.test.ts new file mode 100644 index 000000000..6474c7105 --- /dev/null +++ b/packages/varlock/src/cli/helpers/test/drain-child-output.test.ts @@ -0,0 +1,117 @@ +import { + afterEach, describe, expect, it, vi, +} from 'vitest'; +import { PassThrough } from 'node:stream'; +import { once } from 'node:events'; +import { drainChildOutput, trackChildOutput } from '../drain-child-output'; + +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +describe('drainChildOutput', () => { + it('waits for the actual redacted write callback after the pipe closes', async () => { + let completeWrite: ((error?: Error | null) => void) | undefined; + vi.spyOn(process.stdout, 'write').mockImplementation((_text: any, callback: any) => { + completeWrite = callback; + return false; + }); + const stream = new PassThrough(); + const writer = trackChildOutput(stream, process.stdout); + writer.write('pending redacted output'); + const closed = once(stream, 'close'); + stream.destroy(); + await closed; + let drained = false; + const draining = drainChildOutput({ stdout: stream }).then(() => { + drained = true; + }); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(drained).toBe(false); + completeWrite!(); + await draining; + expect(drained).toBe(true); + }); + + it.each(['EPIPE', 'EIO'])('keeps the existing stderr error policy for %s', async (code) => { + const error = Object.assign(new Error('destination write failed'), { code }); + vi.spyOn(process.stderr, 'write').mockImplementation((_text: any, callback: any) => { + callback(error); + return false; + }); + const stream = new PassThrough(); + trackChildOutput(stream, process.stderr).write('redacted stderr'); + const closed = once(stream, 'close'); + stream.destroy(); + await closed; + const draining = drainChildOutput({ stderr: stream }, { exitCode: 7 }); + if (code === 'EPIPE') await expect(draining).resolves.toBeUndefined(); + else await expect(draining).rejects.toBe(error); + }); + + it('bounds parent writes even after the child pipes close', async () => { + vi.useFakeTimers(); + vi.spyOn(process.stdout, 'write').mockImplementation(() => false); + const stream = new PassThrough(); + trackChildOutput(stream, process.stdout).write('pending redacted output'); + const closed = once(stream, 'close'); + stream.destroy(); + await closed; + const rejection = expect(drainChildOutput({ stdout: stream })).rejects.toThrow('within 2000ms'); + await vi.advanceTimersByTimeAsync(2000); + await rejection; + }); + + it.each([{ exitCode: 7 }, { exitCode: 0, shutdownRequested: true }])('keeps an existing failure or forwarded shutdown status on timeout: %j', async (completion) => { + vi.useFakeTimers(); + const diagnostic = vi.spyOn(console, 'error').mockImplementation(() => {}); + const stream = new PassThrough(); + const draining = drainChildOutput({ stdout: stream }, completion); + await vi.advanceTimersByTimeAsync(2000); + await expect(draining).resolves.toBeUndefined(); + expect(diagnostic).toHaveBeenCalledWith(expect.stringContaining('Output may be incomplete')); + expect(stream.destroyed).toBe(true); + }); + + it('does not wait for inherited terminal streams', async () => { + await expect(drainChildOutput({})).resolves.toBeUndefined(); + }); + + it('accepts pipes that closed before the child exited', async () => { + const stream = new PassThrough(); + const closed = once(stream, 'close'); + stream.destroy(); + await closed; + await expect(drainChildOutput({ stdout: stream })).resolves.toBeUndefined(); + }); + + it('waits until both captured streams close', async () => { + const stdout = new PassThrough(); + const stderr = new PassThrough(); + let drained = false; + const draining = drainChildOutput({ stdout, stderr }).then(() => { + drained = true; + }); + const closed = once(stdout, 'close'); + stdout.destroy(); + await closed; + expect(drained).toBe(false); + stderr.destroy(); + await draining; + expect(drained).toBe(true); + }); + + it('closes pipe readers and rejects rather than hanging past two seconds', async () => { + vi.useFakeTimers(); + const stream = new PassThrough(); + const rejection = expect(drainChildOutput({ stdout: stream })) + .rejects.toThrow('Child output did not finish draining within 2000ms'); + await vi.advanceTimersByTimeAsync(2000); + await rejection; + expect(stream.destroyed).toBe(true); + expect(stream.listenerCount('close')).toBe(0); + }); +}); 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..44a42ef72 --- /dev/null +++ b/smoke-tests/tests/run-output-drain.test.ts @@ -0,0 +1,173 @@ +import { describe, test, expect } from 'vitest'; +import { varlockRun, VARLOCK_CLI } from '../helpers/run-varlock.js'; +import { spawn } from 'node:child_process'; +import { join } from 'node:path'; + +const expected = `${JSON.stringify(['safe-piped-child-output'])}\n`; +const lateWrite = `setTimeout(() => { + process.stdout.write(${JSON.stringify(expected)}); + process.stderr.write('tail:' + process.env.SECRET_TOKEN); +}, 60)`; + +describe('Child output drainage', () => { + test.each(['node', 'bun'])('%s drains real parent writes into a paused reader', async (runtime) => { + const size = 8 * 1024 * 1024; + const code = `process.stdout.write('x'.repeat(${size}) + '\\ntail:' + process.env.SECRET_TOKEN, () => process.exit(0))`; + const result = await new Promise<{ status: number | null; stdout: string; stderr: string }>((resolve, reject) => { + const child = spawn(runtime, [VARLOCK_CLI, 'run', '--', 'node', '-e', code], { + cwd: join(import.meta.dirname, '../smoke-test-basic'), + env: { ...process.env, VARLOCK_TELEMETRY_DISABLED: '1' }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let stdout = ''; + let stderr = ''; + child.stdout.on('data', (chunk) => { + stdout += chunk; + }); + child.stderr.on('data', (chunk) => { + stderr += chunk; + }); + child.stdout.pause(); + const resume = setTimeout(() => { + child.stdout.resume(); + }, 300); + const timer = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error('Paused-reader fixture exceeded its cap')); + }, 10000); + child.on('error', reject); + child.on('close', (status) => { + clearTimeout(timer); + clearTimeout(resume); + resolve({ status, stdout, stderr }); + }); + }); + expect(result.status).toBe(0); + expect(result.stdout.slice(0, size)).toBe('x'.repeat(size)); + expect(result.stdout.slice(size)).toContain('\ntail:'); + expect(result.stdout).not.toContain('super-secret-token-12345'); + expect(result.stderr).toBe(''); + }, 10000); + + test.each(['node', 'bun'])('%s preserves the child status when stderr closes', async (runtime) => { + const code = `process.stderr.write('first\\n', () => setTimeout(() => { + process.stderr.write('second\\n'); + process.stdout.write('intact\\n', () => process.exit(7)); + }, 100))`; + const result = await new Promise<{ + status: number | null; + stdout: string; + closedStderr: boolean; + }>((resolve, reject) => { + const child = spawn(runtime, [VARLOCK_CLI, 'run', '--', 'node', '-e', code], { + cwd: join(import.meta.dirname, '../smoke-test-basic'), + env: { ...process.env, VARLOCK_TELEMETRY_DISABLED: '1' }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let stdout = ''; + let closedStderr = false; + child.stdout.on('data', (chunk) => { + stdout += chunk; + }); + child.stderr.once('data', () => { + closedStderr = true; + child.stderr.destroy(); + }); + const timer = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error('Closed-stderr fixture exceeded its cap')); + }, 10000); + child.on('error', reject); + child.on('close', (status) => { + clearTimeout(timer); + resolve({ status, stdout, closedStderr }); + }); + }); + expect(result.closedStderr).toBe(true); + expect(result.status).toBe(7); + expect(result.stdout).toBe('intact\n'); + }, 10000); + + test.skipIf(process.platform === 'win32')('a signal after child exit cannot make incomplete output successful', async () => { + const holder = 'setTimeout(() => {}, 6000)'; + const code = `require('node:child_process').spawn(process.execPath, ['-e', ${JSON.stringify(holder)}], { + stdio: ['ignore', 1, 2], env: process.env, + }); process.stdout.write('root:' + process.pid + '\\n', () => process.exit(0));`; + const result = await new Promise<{ status: number | null; output: string; sent: boolean }>((resolve, reject) => { + const child = spawn(process.execPath, [VARLOCK_CLI, 'run', '--', 'node', '-e', code], { + cwd: join(import.meta.dirname, '../smoke-test-basic'), + env: { ...process.env, VARLOCK_TELEMETRY_DISABLED: '1' }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let output = ''; + let checking = false; + let sent = false; + const timer = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error('Late-signal fixture exceeded its cap')); + }, 10000); + const receive = async (chunk: Buffer) => { + output += chunk.toString(); + const pid = /root:(\d+)/.exec(output)?.[1]; + if (!pid || checking) return; + checking = true; + for (let attempt = 0; attempt < 100; attempt++) { + try { + process.kill(Number(pid), 0); + } catch (error: any) { + if (error.code === 'ESRCH') { + sent = child.kill('SIGINT'); + return; + } + reject(error); + return; + } + await new Promise((resume) => { + setTimeout(resume, 10); + }); + } + }; + child.stdout.on('data', receive); + child.stderr.on('data', receive); + child.on('error', reject); + child.on('close', (status) => { + clearTimeout(timer); + resolve({ status, output, sent }); + }); + }); + expect(result.sent).toBe(true); + expect(result.status).toBe(1); + expect(result.output).toContain('Child output did not finish draining within 2000ms'); + }, 10000); + + test.each([0, 7])('preserves redacted inherited-pipe output after child exit %i', (exitCode) => { + const child = `require('node:child_process').spawn(process.execPath, ['-e', ${JSON.stringify(lateWrite)}], { + stdio: ['ignore', 1, 2], env: process.env, + }); process.exit(${exitCode});`; + const result = varlockRun(['node', '-e', child], { + cwd: 'smoke-test-basic', + env: { VARLOCK_TELEMETRY_DISABLED: '1' }, + }); + + expect(result.exitCode).toBe(exitCode); + expect(result.stdout).toBe(expected); + expect(result.stderr).toContain('tail:'); + expect(result.output).not.toContain('super-secret-token-12345'); + }); + + test('fails within the drain bound when a descendant keeps a pipe open', () => { + const grandchild = 'setTimeout(() => {}, 6000)'; + const child = `require('node:child_process').spawn(process.execPath, ['-e', ${JSON.stringify(grandchild)}], { + stdio: ['ignore', 1, 2], env: process.env, + }); process.exit(0);`; + const start = performance.now(); + const result = varlockRun(['node', '-e', child], { + cwd: 'smoke-test-basic', + env: { VARLOCK_TELEMETRY_DISABLED: '1' }, + }); + + expect(result.exitCode).not.toBe(0); + expect(result.stderr).toContain('Child output did not finish draining within 2000ms'); + expect(performance.now() - start).toBeLessThan(5000); + }, 10000); +});