ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-27 21:07:01 -06:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
* 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
This commit is contained in:
Alix-007
2026-07-08 12:53:22 +08:00
committed by GitHub
parent 039f8fb16d
commit 35d5ea069a
2 changed files with 178 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/matrix/src/matrix/deps.test.ts
+142 -1
View File
@@ -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", () => {
extensions/matrix/src/matrix/deps.ts
+36 -2
View File
@@ -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,
});
});
});