fix(run): drain redacted child output before exiting

.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.

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.

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;

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<typeof commandSpec> = 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) {

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<object, () => Promise<void>>();
+
+export function trackChildOutput(source: object, destination: Writable) {
+  const pending = new Set<Promise<void>>();
+  let writeError: unknown;
+  outputDrains.set(source, async () => {
+    await Promise.all(pending);
+    if (writeError) throw writeError;
+  });
+  return {
+    write(text: string) {
+      const written = new Promise<void>((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<void> {
+  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<void>((resolve) => {
+    if (stream.closed) {
+      resolve();
+    } else {
+      stream.once('close', resolve);
+      cleanup.push(() => {
+        stream.off('close', resolve);
+      });
+    }
+  }));
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const deadline = new Promise<never>((_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();
+  }
+}

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);
   }

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<void>((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);
+  });
+});