Fix varlock run dropping piped output on exit

.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

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

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

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<typeof commandSpec> = 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 ...');

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<NodeJS.WritableStream>();
+const drainListeners = new WeakSet<NodeJS.WritableStream>();
+
+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<string, string | undefin
   if (streams.length) env[PARENT_REDACTED_STREAMS_ENV_VAR] = `${process.pid}:${streams.join(',')}`;
   else delete env[PARENT_REDACTED_STREAMS_ENV_VAR];
 }
+
+/**
+ * Wait until everything the redaction writers queued on our stdout/stderr has been handed
+ * off. Call before exiting: `gracefulExit` only flushes for up to 1s, which truncates
+ * output when the reader is slower than that (`varlock run -- cmd | slow-consumer`).
+ * Running the command directly would just block on the slow reader, so we do too. A reader
+ * that goes away surfaces as EPIPE, handled by `handleBrokenPipe`; a signal while we wait
+ * exits via the child signal forwarder.
+ */
+export async function flushStdio(): Promise<void> {
+  await Promise.all([process.stdout, process.stderr].map((stream) => new Promise<void>((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();
+  })));
+}

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<NodeJS.Signals> = [];
 
+  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;
     },

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<Readable | null | undefined>, timeoutMs: number): Promise<void> {
+  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<void>((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<ExecResult> & Partial<ExecResult> & { kill: (signal?: number | NodeJS.Signals) => boolean };
 

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<NodeJS.Signals> = ['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;

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<string> = [];
+    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',