fix(matrix): handle stdout/stderr stream errors in dependency commands (#101597)

Alix-007 · web-flow · commit 35d5ea069ae9 · 2026-07-07T21:53:22.000-07:00
* fix(matrix): handle stdout/stderr stream errors in dependency commands

* fix(matrix): type stream-error test process kill

* fix(matrix): harden dependency stream errors

* test(matrix): fix dependency stream test typings
diff --git a/extensions/matrix/src/matrix/deps.test.ts b/extensions/matrix/src/matrix/deps.test.ts
@@ -1,8 +1,9 @@
 // Matrix tests cover deps plugin behavior.
+import type { ChildProcessWithoutNullStreams, SpawnOptions } from "node:child_process";
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
-import { describe, expect, it, vi } from "vitest";
+import { afterEach, describe, expect, it, type MockInstance, vi } from "vitest";
 import {
   ensureMatrixCryptoRuntime,
   ensureMatrixSdkInstalled,
@@ -12,6 +13,72 @@ import {
 
 const logStub = vi.fn();
 
+type ChildKill = (signal?: NodeJS.Signals | number) => boolean;
+
+async function importDepsWithSpawnMock(
+  spawnMock: ReturnType<typeof vi.fn>,
+): Promise<typeof import("./deps.js")> {
+  vi.resetModules();
+  vi.doMock("node:child_process", async () => {
+    const actual = await vi.importActual<typeof import("node:child_process")>("node:child_process");
+    return {
+      ...actual,
+      spawn: spawnMock,
+    };
+  });
+  return await import("./deps.js");
+}
+
+function waitForChildClose(proc: ChildProcessWithoutNullStreams) {
+  return new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
+    const timer = setTimeout(() => {
+      reject(new Error("timed out waiting for matrix command child close"));
+    }, 5_000);
+    timer.unref?.();
+    proc.once("close", (code, signal) => {
+      clearTimeout(timer);
+      resolve({ code, signal });
+    });
+  });
+}
+
+function waitForReadableData(stream: NodeJS.ReadableStream) {
+  return new Promise<void>((resolve, reject) => {
+    let cleanup = () => {};
+    const timer = setTimeout(() => {
+      cleanup();
+      reject(new Error("timed out waiting for matrix command child output"));
+    }, 5_000);
+    timer.unref?.();
+    cleanup = () => {
+      clearTimeout(timer);
+      stream.off("data", onData);
+      stream.off("error", onError);
+    };
+    const onData = () => {
+      cleanup();
+      resolve();
+    };
+    const onError = (error: Error) => {
+      cleanup();
+      reject(error);
+    };
+    stream.once("data", onData);
+    stream.once("error", onError);
+  });
+}
+
+function waitForReadableErrorDispatch() {
+  return new Promise<void>((resolve) => {
+    setImmediate(resolve);
+  });
+}
+
+afterEach(() => {
+  vi.useRealTimers();
+  vi.doUnmock("node:child_process");
+});
+
 function resolveTestNativeBindingFilename(): string | null {
   switch (process.platform) {
     case "darwin":
@@ -185,6 +252,80 @@ describe("runFixedCommandWithTimeout", () => {
     expect(result.stdout).toBe("a".repeat(MATRIX_COMMAND_OUTPUT_TAIL_BYTES));
     expect(result.stderr).toBe("b".repeat(MATRIX_COMMAND_OUTPUT_TAIL_BYTES));
   });
+
+  it("settles real child stream errors after child close and terminates once", async () => {
+    const actual = await vi.importActual<typeof import("node:child_process")>("node:child_process");
+
+    for (const streamName of ["stdout", "stderr"] as const) {
+      let proc: ChildProcessWithoutNullStreams | undefined;
+      let killSpy: MockInstance<ChildKill> | undefined;
+      try {
+        const spawnMock = vi.fn(
+          (command: string, args: string[] | undefined, options: SpawnOptions) => {
+            proc = actual.spawn(command, args ?? [], options) as ChildProcessWithoutNullStreams;
+            return proc;
+          },
+        );
+        const { runFixedCommandWithTimeout: runWithMockedSpawn } =
+          await importDepsWithSpawnMock(spawnMock);
+        const exitListenersBefore = process.listenerCount("exit");
+
+        const resultPromise = runWithMockedSpawn({
+          argv: [
+            process.execPath,
+            "-e",
+            [
+              "process.stdin.resume();",
+              'process.on("SIGTERM", () => {});',
+              'process.stdout.write("stdout ready\\n");',
+              'process.stderr.write("stderr ready\\n");',
+              "setInterval(() => {}, 1000);",
+            ].join(""),
+          ],
+          cwd: process.cwd(),
+          timeoutMs: 10_000,
+        });
+        if (!proc) {
+          throw new Error("expected matrix command helper to spawn a child process");
+        }
+        killSpy = vi.spyOn(proc, "kill");
+        const closePromise = waitForChildClose(proc);
+        await Promise.all([waitForReadableData(proc.stdout), waitForReadableData(proc.stderr)]);
+        let settled = false;
+        void resultPromise.then(() => {
+          settled = true;
+        });
+        const message = `synthetic parent ${streamName} read failure`;
+
+        proc[streamName].destroy(new Error(message));
+        await waitForReadableErrorDispatch();
+        expect(settled).toBe(false);
+        expect(process.listenerCount("exit")).toBe(exitListenersBefore + 1);
+        expect(killSpy).toHaveBeenCalledTimes(1);
+        expect(killSpy).toHaveBeenCalledWith("SIGTERM");
+
+        const duplicateStreamName = streamName === "stdout" ? "stderr" : "stdout";
+        proc[duplicateStreamName].destroy(new Error("duplicate parent readable failure"));
+        await waitForReadableErrorDispatch();
+        expect(killSpy).toHaveBeenCalledTimes(1);
+
+        const result = await resultPromise;
+        const close = await closePromise;
+
+        expect(result.code).toBe(1);
+        expect(result.stderr).toContain(`${streamName} stream failed: ${message}`);
+        expect(result.stderr).not.toContain("duplicate parent readable failure");
+        expect(close).toStrictEqual({ code: null, signal: "SIGKILL" });
+        expect(killSpy).toHaveBeenLastCalledWith("SIGKILL");
+        expect(process.listenerCount("exit")).toBe(exitListenersBefore);
+      } finally {
+        killSpy?.mockRestore();
+        if (proc && proc.exitCode === null && !proc.killed) {
+          proc.kill("SIGKILL");
+        }
+      }
+    }
+  });
 });
 
 describe("ensureMatrixSdkInstalled", () => {
diff --git a/extensions/matrix/src/matrix/deps.ts b/extensions/matrix/src/matrix/deps.ts
@@ -13,6 +13,7 @@ const REQUIRED_MATRIX_PACKAGES = [
 ];
 const MIN_MATRIX_CRYPTO_NATIVE_BINDING_BYTES = 1_000_000;
 export const MATRIX_COMMAND_OUTPUT_TAIL_BYTES = 64 * 1024;
+const MATRIX_STREAM_ERROR_KILL_GRACE_MS = 1_000;
 
 type MatrixCryptoRuntimeDeps = {
   requireFn?: (id: string) => unknown;
@@ -99,6 +100,8 @@ export async function runFixedCommandWithTimeout(params: {
     let stderr = "";
     let settled = false;
     let timer: NodeJS.Timeout | null = null;
+    let streamKillTimer: NodeJS.Timeout | null = null;
+    let streamErrorMessage: string | null = null;
     const killChildOnExit = () => {
       if (!settled && proc.exitCode === null) {
         proc.kill("SIGTERM");
@@ -113,6 +116,9 @@ export async function runFixedCommandWithTimeout(params: {
       if (timer) {
         clearTimeout(timer);
       }
+      if (streamKillTimer) {
+        clearTimeout(streamKillTimer);
+      }
       process.off("exit", killChildOnExit);
       resolve(result);
     };
@@ -124,9 +130,29 @@ export async function runFixedCommandWithTimeout(params: {
     proc.stderr?.on("data", (chunk: Buffer | string) => {
       stderr = appendBoundedOutputTail(stderr, chunk);
     });
+    const failReadableStream = (streamName: "stdout" | "stderr") => (error: Error) => {
+      if (settled || streamErrorMessage) {
+        return;
+      }
+      streamErrorMessage = `${streamName} stream failed: ${formatErrorMessage(error)}`;
+      if (proc.exitCode === null) {
+        proc.kill("SIGTERM");
+      }
+      streamKillTimer = setTimeout(() => {
+        if (!settled && proc.exitCode === null) {
+          proc.kill("SIGKILL");
+        }
+      }, MATRIX_STREAM_ERROR_KILL_GRACE_MS);
+      streamKillTimer.unref?.();
+    };
+    proc.stdout?.on("error", failReadableStream("stdout"));
+    proc.stderr?.on("error", failReadableStream("stderr"));
 
     timer = setTimeout(() => {
       proc.kill("SIGKILL");
+      if (streamErrorMessage) {
+        return;
+      }
       finalize({
         code: 124,
         stdout,
@@ -135,6 +161,9 @@ export async function runFixedCommandWithTimeout(params: {
     }, params.timeoutMs);
 
     proc.on("error", (err) => {
+      if (streamErrorMessage) {
+        return;
+      }
       finalize({
         code: 1,
         stdout,
@@ -143,10 +172,15 @@ export async function runFixedCommandWithTimeout(params: {
     });
 
     proc.on("close", (code) => {
+      const streamErrorStderr = streamErrorMessage
+        ? stderr
+          ? appendBoundedOutputTail(stderr, `\n${streamErrorMessage}`)
+          : streamErrorMessage
+        : stderr;
       finalize({
-        code: code ?? 1,
+        code: streamErrorMessage ? 1 : (code ?? 1),
         stdout,
-        stderr,
+        stderr: streamErrorStderr,
       });
     });
   });
