From a1d31204f120e4f9d87b631a70a52ec53389022b Mon Sep 17 00:00:00 2001 From: wm0018 Date: Wed, 29 Jul 2026 14:28:14 +0800 Subject: [PATCH] fix(gateway): drain root work continuations before direct-stop process exit (#105848) Co-authored-by: Peter Steinberger --- src/cli/gateway-cli/run-loop.test.ts | 142 ++++++++++++++++++ src/cli/gateway-cli/run-loop.ts | 25 ++- test/gateway-session-end-shutdown.e2e.test.ts | 139 +++++++++++++++++ 3 files changed, 303 insertions(+), 3 deletions(-) create mode 100644 test/gateway-session-end-shutdown.e2e.test.ts diff --git a/src/cli/gateway-cli/run-loop.test.ts b/src/cli/gateway-cli/run-loop.test.ts index 7dd0b30cf121..79ca9066f195 100644 --- a/src/cli/gateway-cli/run-loop.test.ts +++ b/src/cli/gateway-cli/run-loop.test.ts @@ -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((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((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(); diff --git a/src/cli/gateway-cli/run-loop.ts b/src/cli/gateway-cli/run-loop.ts index 1efd0b2917ab..0e346a87fc87 100644 --- a/src/cli/gateway-cli/run-loop.ts +++ b/src/cli/gateway-cli/run-loop.ts @@ -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) { diff --git a/test/gateway-session-end-shutdown.e2e.test.ts b/test/gateway-session-end-shutdown.e2e.test.ts new file mode 100644 index 000000000000..766ad3d09bf2 --- /dev/null +++ b/test/gateway-session-end-shutdown.e2e.test.ts @@ -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 { + 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 { + 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, + ); +});