mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
fix(gateway): drain root work continuations before direct-stop process exit (#105848)
Co-authored-by: Peter Steinberger <steipete@gmail.com>
This commit is contained in:
wm0018
@@ -499,6 +499,148 @@ describe("runGatewayLoop", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["SIGTERM", "SIGINT"] as const)(
|
||||
"drains admitted root work before closing on %s",
|
||||
async (signal) => {
|
||||
vi.clearAllMocks();
|
||||
|
||||
await withIsolatedSignals(async ({ captureSignal }) => {
|
||||
const { close, runtime, exited } = await createSignaledLoopHarness();
|
||||
let releaseDrain: (() => void) | undefined;
|
||||
const pendingDrain = new Promise<void>((resolve) => {
|
||||
releaseDrain = resolve;
|
||||
});
|
||||
waitForActiveGatewayRootWork.mockImplementationOnce(async () => {
|
||||
await pendingDrain;
|
||||
return { drained: true, active: 0 };
|
||||
});
|
||||
|
||||
try {
|
||||
captureSignal(signal)();
|
||||
await waitForLoopCondition(
|
||||
() => waitForActiveGatewayRootWork.mock.calls.length === 1,
|
||||
`expected ${signal} to drain admitted gateway root work`,
|
||||
);
|
||||
|
||||
expect(markGatewayDraining).toHaveBeenCalledOnce();
|
||||
expect(markGatewayDraining.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
waitForActiveGatewayRootWork.mock.invocationCallOrder[0] ?? 0,
|
||||
);
|
||||
expect(waitForActiveGatewayRootWork).toHaveBeenCalledWith(15_000);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(runtime.exit).not.toHaveBeenCalled();
|
||||
|
||||
releaseDrain?.();
|
||||
|
||||
await expect(exited).resolves.toBe(0);
|
||||
expect(close).toHaveBeenCalledWith({
|
||||
reason: "gateway stopping",
|
||||
restartExpectedMs: null,
|
||||
});
|
||||
} finally {
|
||||
releaseDrain?.();
|
||||
await exited;
|
||||
waitForActiveGatewayRootWork.mockReset();
|
||||
waitForActiveGatewayRootWork.mockResolvedValue({ drained: true, active: 0 });
|
||||
}
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it("continues direct shutdown when the bounded root-work drain times out", async () => {
|
||||
vi.clearAllMocks();
|
||||
|
||||
await withIsolatedSignals(async ({ captureSignal }) => {
|
||||
waitForActiveGatewayRootWork.mockResolvedValueOnce({ drained: false, active: 2 });
|
||||
const { close, runtime, exited } = await createSignaledLoopHarness();
|
||||
|
||||
try {
|
||||
captureSignal("SIGTERM")();
|
||||
|
||||
await expect(exited).resolves.toBe(0);
|
||||
expect(waitForActiveGatewayRootWork).toHaveBeenCalledWith(15_000);
|
||||
expect(gatewayLog.warn).toHaveBeenCalledWith(
|
||||
"gateway root transaction drain timeout reached with 2 root(s) still active; proceeding with shutdown",
|
||||
);
|
||||
expect(close).toHaveBeenCalledWith({
|
||||
reason: "gateway stopping",
|
||||
restartExpectedMs: null,
|
||||
});
|
||||
expect(runtime.exit).toHaveBeenCalledWith(0);
|
||||
} finally {
|
||||
waitForActiveGatewayRootWork.mockReset();
|
||||
waitForActiveGatewayRootWork.mockResolvedValue({ drained: true, active: 0 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("still closes and exits when the direct-shutdown root-work drain fails", async () => {
|
||||
vi.clearAllMocks();
|
||||
|
||||
await withIsolatedSignals(async ({ captureSignal }) => {
|
||||
waitForActiveGatewayRootWork.mockRejectedValueOnce(new Error("root drain unavailable"));
|
||||
const { close, runtime, exited } = await createSignaledLoopHarness();
|
||||
|
||||
try {
|
||||
captureSignal("SIGTERM")();
|
||||
|
||||
await expect(exited).resolves.toBe(0);
|
||||
expect(waitForActiveGatewayRootWork).toHaveBeenCalledWith(15_000);
|
||||
expect(gatewayLog.warn).toHaveBeenCalledWith(
|
||||
"gateway root transaction drain failed; proceeding with shutdown: root drain unavailable",
|
||||
);
|
||||
expect(close).toHaveBeenCalledWith({
|
||||
reason: "gateway stopping",
|
||||
restartExpectedMs: null,
|
||||
});
|
||||
expect(runtime.exit).toHaveBeenCalledWith(0);
|
||||
} finally {
|
||||
waitForActiveGatewayRootWork.mockReset();
|
||||
waitForActiveGatewayRootWork.mockResolvedValue({ drained: true, active: 0 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("does not start a second root-work drain for repeated shutdown signals", async () => {
|
||||
vi.clearAllMocks();
|
||||
|
||||
await withIsolatedSignals(async ({ captureSignal }) => {
|
||||
const { exited } = await createSignaledLoopHarness();
|
||||
let releaseDrain: (() => void) | undefined;
|
||||
const pendingDrain = new Promise<void>((resolve) => {
|
||||
releaseDrain = resolve;
|
||||
});
|
||||
waitForActiveGatewayRootWork.mockImplementationOnce(async () => {
|
||||
await pendingDrain;
|
||||
return { drained: true, active: 0 };
|
||||
});
|
||||
|
||||
try {
|
||||
const sigterm = captureSignal("SIGTERM");
|
||||
const sigint = captureSignal("SIGINT");
|
||||
sigterm();
|
||||
await waitForLoopCondition(
|
||||
() => waitForActiveGatewayRootWork.mock.calls.length === 1,
|
||||
"expected first shutdown signal to begin the root-work drain",
|
||||
);
|
||||
|
||||
sigint();
|
||||
|
||||
expect(waitForActiveGatewayRootWork).toHaveBeenCalledOnce();
|
||||
expect(markGatewayDraining).toHaveBeenCalledOnce();
|
||||
expect(gatewayLog.info).toHaveBeenCalledWith("received SIGINT during shutdown; ignoring");
|
||||
|
||||
releaseDrain?.();
|
||||
await expect(exited).resolves.toBe(0);
|
||||
} finally {
|
||||
releaseDrain?.();
|
||||
await exited;
|
||||
waitForActiveGatewayRootWork.mockReset();
|
||||
waitForActiveGatewayRootWork.mockResolvedValue({ drained: true, active: 0 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("bounds the file-log flush before a graceful SIGTERM exit", async () => {
|
||||
vi.clearAllMocks();
|
||||
|
||||
|
||||
@@ -746,6 +746,25 @@ export async function runGatewayLoop(params: {
|
||||
);
|
||||
}
|
||||
|
||||
if (!isRestart) {
|
||||
// Keep reset-started finalizers alive without spending the shutdown
|
||||
// reserve that server teardown and the supervisor watchdog need.
|
||||
try {
|
||||
const rootDrain = await eagerLifecycleRuntime.waitForActiveGatewayRootWork(
|
||||
Math.max(0, SHUTDOWN_TIMEOUT_MS - RESTART_CLOSE_REPLY_DRAIN_SHUTDOWN_RESERVE_MS),
|
||||
);
|
||||
if (!rootDrain.drained) {
|
||||
gatewayLog.warn(
|
||||
`gateway root transaction drain timeout reached with ${rootDrain.active} root(s) still active; proceeding with shutdown`,
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
gatewayLog.warn(
|
||||
`gateway root transaction drain failed; proceeding with shutdown: ${formatErrorMessage(err)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
armCloseForceExitTimerForIndefiniteRestart();
|
||||
const closeDrainTimeoutMs = resolveRestartCloseDrainTimeoutMs();
|
||||
await server?.close({
|
||||
@@ -831,9 +850,9 @@ export async function runGatewayLoop(params: {
|
||||
return;
|
||||
}
|
||||
const isRestart = action === "restart";
|
||||
if (isRestart) {
|
||||
markRestartDraining();
|
||||
}
|
||||
// Fence new roots synchronously for stops as well as restarts so admitted
|
||||
// detached finalizers can drain before the signal tears down the gateway.
|
||||
markRestartDraining();
|
||||
shuttingDown = true;
|
||||
gatewayLog.info(`received ${signal}; ${isRestart ? "restarting" : "shutting down"}`);
|
||||
if (isRestart) {
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
// A real Gateway process must finish reset-started plugin work before SIGTERM exit.
|
||||
import { once } from "node:events";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { OpenClawConfig } from "../src/config/types.openclaw.js";
|
||||
import { connectGatewayClient, disconnectGatewayClient } from "../src/gateway/test-helpers.e2e.js";
|
||||
import {
|
||||
createOpenClawTestInstance,
|
||||
type OpenClawTestInstance,
|
||||
} from "./helpers/openclaw-test-instance.js";
|
||||
|
||||
const PLUGIN_ID = "session-end-shutdown-proof";
|
||||
const SESSION_KEY = "agent:main:dashboard:session-end-shutdown-proof";
|
||||
const HOOK_DELAY_MS = 10_000;
|
||||
const TEST_TIMEOUT_MS = 120_000;
|
||||
const WAIT_OPTIONS = { timeout: 10_000, interval: 25 } as const;
|
||||
|
||||
const instances: OpenClawTestInstance[] = [];
|
||||
const fixtureDirs: string[] = [];
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(instances.splice(0).map(async (instance) => await instance.cleanup()));
|
||||
await Promise.all(
|
||||
fixtureDirs.splice(0).map(async (dir) => await rm(dir, { recursive: true, force: true })),
|
||||
);
|
||||
});
|
||||
|
||||
async function writeSessionEndPlugin(pluginDir: string, tracePath: string): Promise<void> {
|
||||
await mkdir(pluginDir, { recursive: true });
|
||||
await writeFile(
|
||||
path.join(pluginDir, "openclaw.plugin.json"),
|
||||
`${JSON.stringify({
|
||||
id: PLUGIN_ID,
|
||||
name: "Session End Shutdown Proof",
|
||||
activation: { onStartup: true },
|
||||
configSchema: { type: "object", additionalProperties: false, properties: {} },
|
||||
})}\n`,
|
||||
);
|
||||
await writeFile(
|
||||
path.join(pluginDir, "index.mjs"),
|
||||
[
|
||||
'import { appendFileSync } from "node:fs";',
|
||||
"export default {",
|
||||
` id: ${JSON.stringify(PLUGIN_ID)},`,
|
||||
" register(api) {",
|
||||
' api.on("session_end", async (event) => {',
|
||||
` if (event.sessionKey !== ${JSON.stringify(SESSION_KEY)} || event.reason !== "reset") return;`,
|
||||
` appendFileSync(${JSON.stringify(tracePath)}, "started\\n");`,
|
||||
` await new Promise((resolve) => setTimeout(resolve, ${HOOK_DELAY_MS}));`,
|
||||
` appendFileSync(${JSON.stringify(tracePath)}, "completed\\n");`,
|
||||
" });",
|
||||
" },",
|
||||
"};",
|
||||
"",
|
||||
].join("\n"),
|
||||
);
|
||||
}
|
||||
|
||||
async function readTrace(tracePath: string): Promise<string[]> {
|
||||
try {
|
||||
return (await readFile(tracePath, "utf8")).split("\n").filter(Boolean);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
|
||||
return [];
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
describe("Gateway session-end shutdown", () => {
|
||||
it(
|
||||
"finishes a real reset-started session_end hook after an operating-system SIGTERM",
|
||||
async () => {
|
||||
const fixtureDir = await mkdtemp(path.join(tmpdir(), "openclaw-session-end-shutdown-"));
|
||||
fixtureDirs.push(fixtureDir);
|
||||
const pluginDir = path.join(fixtureDir, "plugin");
|
||||
const tracePath = path.join(fixtureDir, "session-end.trace");
|
||||
await writeSessionEndPlugin(pluginDir, tracePath);
|
||||
|
||||
const config = {
|
||||
plugins: {
|
||||
enabled: true,
|
||||
allow: [PLUGIN_ID],
|
||||
load: { paths: [pluginDir] },
|
||||
entries: { [PLUGIN_ID]: { enabled: true } },
|
||||
slots: { memory: "none" },
|
||||
},
|
||||
} satisfies OpenClawConfig;
|
||||
const instance = await createOpenClawTestInstance({
|
||||
name: "session-end-shutdown",
|
||||
config,
|
||||
env: { OPENCLAW_TEST_MINIMAL_GATEWAY: undefined },
|
||||
stopTimeoutMs: 10_000,
|
||||
});
|
||||
instances.push(instance);
|
||||
await instance.startGateway();
|
||||
|
||||
const client = await connectGatewayClient({
|
||||
url: instance.url,
|
||||
token: instance.gatewayToken,
|
||||
role: "operator",
|
||||
scopes: ["operator.admin", "operator.read", "operator.write"],
|
||||
});
|
||||
|
||||
try {
|
||||
await vi.waitFor(async () => {
|
||||
const created = await client.request<{ key: string; sessionId: string }>(
|
||||
"sessions.create",
|
||||
{ agentId: "main", key: SESSION_KEY },
|
||||
);
|
||||
expect(created.key).toBe(SESSION_KEY);
|
||||
expect(created.sessionId).toBeTruthy();
|
||||
}, WAIT_OPTIONS);
|
||||
|
||||
await client.request("sessions.reset", { key: SESSION_KEY, reason: "reset" });
|
||||
await vi.waitFor(async () => {
|
||||
expect(await readTrace(tracePath), instance.logs()).toEqual(["started"]);
|
||||
}, WAIT_OPTIONS);
|
||||
|
||||
const child = instance.child;
|
||||
if (!child) {
|
||||
throw new Error("Gateway process exited before its session-end hook was signaled");
|
||||
}
|
||||
const exited = once(child, "exit") as Promise<
|
||||
[code: number | null, signal: NodeJS.Signals | null]
|
||||
>;
|
||||
expect(child.kill("SIGTERM")).toBe(true);
|
||||
|
||||
await expect(exited).resolves.toEqual([0, null]);
|
||||
await expect(readTrace(tracePath)).resolves.toEqual(["started", "completed"]);
|
||||
} finally {
|
||||
await disconnectGatewayClient(client).catch(() => undefined);
|
||||
}
|
||||
},
|
||||
TEST_TIMEOUT_MS,
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user