From 35d5ea069ae90d9377df69dbdb5c636c05f58ecd Mon Sep 17 00:00:00 2001 From: Alix-007 Date: Wed, 8 Jul 2026 12:53:22 +0800 Subject: [PATCH] fix(matrix): handle stdout/stderr stream errors in dependency commands (#101597) * 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 --- extensions/matrix/src/matrix/deps.test.ts | 143 +++++++++++++++++++++- extensions/matrix/src/matrix/deps.ts | 38 +++++- 2 files changed, 178 insertions(+), 3 deletions(-) diff --git a/extensions/matrix/src/matrix/deps.test.ts b/extensions/matrix/src/matrix/deps.test.ts index 3f504c6c9d24..3a190082116b 100644 --- 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, +): Promise { + vi.resetModules(); + vi.doMock("node:child_process", async () => { + const actual = await vi.importActual("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((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((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("node:child_process"); + + for (const streamName of ["stdout", "stderr"] as const) { + let proc: ChildProcessWithoutNullStreams | undefined; + let killSpy: MockInstance | 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 index 0c72fa06dcf2..0b0cdb1fa101 100644 --- 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, }); }); });