From 00d3f931c996824701bf1ebed5f67f4822d86ef6 Mon Sep 17 00:00:00 2001 From: Honey Tyagi Date: Wed, 8 Jul 2026 22:38:48 +0530 Subject: [PATCH] fix(broker): self-terminate on idle to reap orphaned shared brokers (#450) A shared/worktree broker records its owning sessionIds in broker.json and is torn down only once no owner remains. If a co-owning session disappears without running its SessionEnd hook (SIGKILL, OOM, crash, host reboot) or its teardown skips the entry on lock contention, its sessionId lingers forever and no future hook fires for it, orphaning the broker indefinitely. Fix it broker-side: app-server-broker.mjs now self-terminates after an idle timeout with no connected client. This is platform-independent, needs no PID/liveness signal, and covers the abnormal-exit orphan, the dead-co-owner orphan, and the lock-contention skip in one mechanism (see #108, #380, #450). The timeout is configurable via --idle-timeout or CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS (default 30m); a value <= 0 disables it. The timer arms on listen and whenever the last client disconnects, and disarms while a client is connected. --- plugins/codex/scripts/app-server-broker.mjs | 69 +++++++++- tests/broker-idle-timeout.test.mjs | 136 ++++++++++++++++++++ 2 files changed, 202 insertions(+), 3 deletions(-) create mode 100644 tests/broker-idle-timeout.test.mjs diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1954274f..e62db254 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -11,6 +11,34 @@ import { parseBrokerEndpoint } from "./lib/broker-endpoint.mjs"; const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]); +// Broker-side idle timeout. When no client has been connected for this long the +// broker self-terminates. This is the correctness backstop for stale ownership +// records in broker.json: a co-owning session that exits without running its +// SessionEnd hook (SIGKILL, OOM, crash, host reboot) or whose teardown skips the +// entry on lock contention leaves its sessionId behind forever, so no future hook +// will ever tear the broker down. Self-termination on idle is platform-independent +// and needs no PID/liveness signal, so it covers the abnormal-exit orphan, the +// dead-co-owner orphan, and the lock-contention skip in one mechanism. See #108, +// #380, and #450. +const IDLE_TIMEOUT_ENV = "CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS"; +const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; + +// Resolve the idle timeout from the CLI flag, then the environment, then the +// default. A value <= 0 (or a non-finite/negative override) disables the timeout +// so the broker never self-terminates, which keeps the old behavior available for +// callers that manage lifecycle themselves. +function resolveIdleTimeoutMs(optionValue, env = process.env) { + const raw = optionValue ?? env[IDLE_TIMEOUT_ENV]; + if (raw === undefined || raw === null || raw === "") { + return DEFAULT_IDLE_TIMEOUT_MS; + } + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed < 0) { + return DEFAULT_IDLE_TIMEOUT_MS; + } + return parsed; +} + function buildStreamThreadIds(method, params, result) { const threadIds = new Set(); if (params?.threadId) { @@ -48,11 +76,11 @@ function writePidFile(pidFile) { async function main() { const [subcommand, ...argv] = process.argv.slice(2); if (subcommand !== "serve") { - throw new Error("Usage: node scripts/app-server-broker.mjs serve --endpoint [--cwd ] [--pid-file ]"); + throw new Error("Usage: node scripts/app-server-broker.mjs serve --endpoint [--cwd ] [--pid-file ] [--idle-timeout ]"); } const { options } = parseArgs(argv, { - valueOptions: ["cwd", "pid-file", "endpoint"] + valueOptions: ["cwd", "pid-file", "endpoint", "idle-timeout"] }); if (!options.endpoint) { @@ -63,6 +91,7 @@ async function main() { const endpoint = String(options.endpoint); const listenTarget = parseBrokerEndpoint(endpoint); const pidFile = options["pid-file"] ? path.resolve(options["pid-file"]) : null; + const idleTimeoutMs = resolveIdleTimeoutMs(options["idle-timeout"]); writePidFile(pidFile); const appClient = await CodexAppServerClient.connect(cwd, { disableBroker: true }); @@ -70,6 +99,31 @@ async function main() { let activeStreamSocket = null; let activeStreamThreadIds = null; const sockets = new Set(); + let idleTimer = null; + + function disarmIdleTimer() { + if (idleTimer) { + clearTimeout(idleTimer); + idleTimer = null; + } + } + + // Arm the idle timer whenever the broker has no connected clients. A live + // client connection means the broker is still in use, so we only count down + // while idle and cancel the moment a client connects. When the timer fires we + // shut the broker down gracefully and exit. + function armIdleTimer() { + disarmIdleTimer(); + if (idleTimeoutMs <= 0 || sockets.size > 0) { + return; + } + idleTimer = setTimeout(() => { + idleTimer = null; + shutdown(server) + .catch(() => {}) + .finally(() => process.exit(0)); + }, idleTimeoutMs); + } function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { @@ -100,6 +154,7 @@ async function main() { } async function shutdown(server) { + disarmIdleTimer(); for (const socket of sockets) { socket.end(); } @@ -117,6 +172,7 @@ async function main() { const server = net.createServer((socket) => { sockets.add(socket); + disarmIdleTimer(); socket.setEncoding("utf8"); let buffer = ""; @@ -225,11 +281,13 @@ async function main() { socket.on("close", () => { sockets.delete(socket); clearSocketOwnership(socket); + armIdleTimer(); }); socket.on("error", () => { sockets.delete(socket); clearSocketOwnership(socket); + armIdleTimer(); }); }); @@ -243,7 +301,12 @@ async function main() { process.exit(0); }); - server.listen(listenTarget.path); + server.listen(listenTarget.path, () => { + // Start counting down immediately: a broker that is spawned but never + // receives a client (or whose only client connects briefly during the + // readiness probe) must still self-terminate instead of lingering. + armIdleTimer(); + }); } main().catch((error) => { diff --git a/tests/broker-idle-timeout.test.mjs b/tests/broker-idle-timeout.test.mjs new file mode 100644 index 00000000..4a848897 --- /dev/null +++ b/tests/broker-idle-timeout.test.mjs @@ -0,0 +1,136 @@ +import fs from "node:fs"; +import net from "node:net"; +import path from "node:path"; +import test from "node:test"; +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; + +import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; +import { makeTempDir } from "./helpers.mjs"; +import { createBrokerEndpoint, parseBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; +import { waitForBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; + +const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const BROKER_SCRIPT = path.join(ROOT, "plugins", "codex", "scripts", "app-server-broker.mjs"); + +function spawnBroker({ cwd, endpoint, env, idleTimeoutMs }) { + const args = [BROKER_SCRIPT, "serve", "--endpoint", endpoint, "--cwd", cwd]; + if (idleTimeoutMs !== undefined) { + args.push("--idle-timeout", String(idleTimeoutMs)); + } + return spawn(process.execPath, args, { + cwd, + env, + stdio: ["ignore", "pipe", "pipe"] + }); +} + +function waitForExit(child, { timeoutMs = 5000 } = {}) { + return new Promise((resolve, reject) => { + if (child.exitCode !== null || child.signalCode !== null) { + resolve({ code: child.exitCode, signal: child.signalCode }); + return; + } + const timer = setTimeout(() => { + child.removeListener("exit", onExit); + reject(new Error("Timed out waiting for broker process to exit.")); + }, timeoutMs); + function onExit(code, signal) { + clearTimeout(timer); + resolve({ code, signal }); + } + child.once("exit", onExit); + }); +} + +function connectClient(endpoint) { + const target = parseBrokerEndpoint(endpoint); + return new Promise((resolve, reject) => { + const socket = net.createConnection({ path: target.path }); + socket.once("connect", () => resolve(socket)); + socket.once("error", reject); + }); +} + +function delay(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +test("broker self-terminates after the idle timeout when no client is connected", async () => { + const binDir = makeTempDir(); + installFakeCodex(binDir); + const sessionDir = makeTempDir("cxc-"); + const endpoint = createBrokerEndpoint(sessionDir); + const child = spawnBroker({ cwd: sessionDir, endpoint, env: buildEnv(binDir), idleTimeoutMs: 300 }); + + try { + const ready = await waitForBrokerEndpoint(endpoint, 3000); + assert.equal(ready, true, "broker should accept connections before it times out"); + + const start = Date.now(); + const result = await waitForExit(child, { timeoutMs: 5000 }); + const elapsed = Date.now() - start; + + assert.equal(result.code, 0, "broker should exit cleanly on idle timeout"); + assert.ok(elapsed >= 200, `broker exited too early (${elapsed}ms)`); + } finally { + if (child.exitCode === null && child.signalCode === null) { + child.kill(); + } + } +}); + +test("broker stays alive while a client is connected and exits after it disconnects", async () => { + const binDir = makeTempDir(); + installFakeCodex(binDir); + const sessionDir = makeTempDir("cxc-"); + const endpoint = createBrokerEndpoint(sessionDir); + const child = spawnBroker({ cwd: sessionDir, endpoint, env: buildEnv(binDir), idleTimeoutMs: 300 }); + + let socket = null; + try { + const ready = await waitForBrokerEndpoint(endpoint, 3000); + assert.equal(ready, true); + + socket = await connectClient(endpoint); + + // Hold the connection open well past the idle timeout; the broker must not + // self-terminate while a client is still connected. + await delay(900); + assert.equal(child.exitCode, null, "broker must stay alive while a client is connected"); + + socket.end(); + socket = null; + + const result = await waitForExit(child, { timeoutMs: 5000 }); + assert.equal(result.code, 0, "broker should exit once the client disconnects and it goes idle"); + } finally { + if (socket) { + socket.destroy(); + } + if (child.exitCode === null && child.signalCode === null) { + child.kill(); + } + } +}); + +test("broker with the idle timeout disabled keeps running while idle", async () => { + const binDir = makeTempDir(); + installFakeCodex(binDir); + const sessionDir = makeTempDir("cxc-"); + const endpoint = createBrokerEndpoint(sessionDir); + const child = spawnBroker({ cwd: sessionDir, endpoint, env: buildEnv(binDir), idleTimeoutMs: 0 }); + + try { + const ready = await waitForBrokerEndpoint(endpoint, 3000); + assert.equal(ready, true); + + await delay(700); + assert.equal(child.exitCode, null, "broker must not self-terminate when the idle timeout is disabled"); + } finally { + if (child.exitCode === null && child.signalCode === null) { + child.kill(); + } + } +});