diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs index ef763819c..a96595c69 100644 --- a/plugins/codex/scripts/lib/broker-lifecycle.mjs +++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs @@ -99,23 +99,108 @@ export function clearBrokerSession(cwd) { } } +// A broker that is mid-turn keeps its event loop busy, so 150 ms is not enough +// budget for it to accept a probe connection. Failing the probe used to be treated +// as "the broker is gone", which then tore down a perfectly healthy broker. +const BROKER_READY_PROBE_TIMEOUT_MS = 3000; + +// Stale-lock timeout for the broker creation lock below. +const BROKER_LOCK_STALE_MS = 30000; +const BROKER_LOCK_WAIT_MS = 15000; + async function isBrokerEndpointReady(endpoint) { if (!endpoint) { return false; } try { - return await waitForBrokerEndpoint(endpoint, 150); + return await waitForBrokerEndpoint(endpoint, BROKER_READY_PROBE_TIMEOUT_MS); } catch { return false; } } +// A failed probe only proves the broker did not answer in time; it does not prove +// the process is gone. Never tear down a broker whose process is still alive. +function isBrokerProcessAlive(pid) { + if (!Number.isFinite(pid) || pid <= 0) { + return false; + } + try { + process.kill(pid, 0); + return true; + } catch (error) { + // ESRCH is the only code that proves the process is gone. EPERM (and anything + // else) means we cannot tell, so assume it is alive and leave it alone. + return error?.code !== "ESRCH"; + } +} + +function brokerLockPath(cwd) { + return path.join(resolveStateDir(cwd), "broker.lock"); +} + +// Concurrent clients can all observe `loadBrokerSession() === null` and each spawn a +// broker; the last writer wins and the other brokers are orphaned together with the +// app-servers (and in-flight turns) they own. Serialize the check-then-create window. +async function withBrokerLock(cwd, fn) { + const lockFile = brokerLockPath(cwd); + fs.mkdirSync(path.dirname(lockFile), { recursive: true }); + + const deadline = Date.now() + BROKER_LOCK_WAIT_MS; + let held = false; + while (Date.now() < deadline) { + try { + const fd = fs.openSync(lockFile, "wx"); + fs.writeSync(fd, String(process.pid)); + fs.closeSync(fd); + held = true; + break; + } catch (error) { + if (error?.code !== "EEXIST") { + throw error; + } + try { + if (Date.now() - fs.statSync(lockFile).mtimeMs > BROKER_LOCK_STALE_MS) { + fs.unlinkSync(lockFile); + continue; + } + } catch { + // The holder released it between statSync and now; retry. + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + } + + try { + // Falling through without the lock keeps the previous (racy) behaviour rather + // than failing the caller outright. + return await fn(); + } finally { + if (held) { + try { + fs.unlinkSync(lockFile); + } catch { + // Already removed by a stale-lock sweep. + } + } + } +} + export async function ensureBrokerSession(cwd, options = {}) { + return withBrokerLock(cwd, () => ensureBrokerSessionLocked(cwd, options)); +} + +async function ensureBrokerSessionLocked(cwd, options = {}) { const existing = loadBrokerSession(cwd); if (existing && (await isBrokerEndpointReady(existing.endpoint))) { return existing; } + if (existing && isBrokerProcessAlive(existing.pid)) { + // Alive but slow to answer: it is busy, not dead. Reuse it. + return existing; + } + if (existing) { teardownBrokerSession({ endpoint: existing.endpoint ?? null, diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs new file mode 100644 index 000000000..3bf2883cf --- /dev/null +++ b/tests/broker-lifecycle.test.mjs @@ -0,0 +1,83 @@ +import fs from "node:fs"; +import path from "node:path"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { makeTempDir } from "./helpers.mjs"; +import { ensureBrokerSession, saveBrokerSession, loadBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; + +test("ensureBrokerSession reuses a live broker whose endpoint does not answer the probe", async () => { + const workspace = makeTempDir(); + // Use this test process as the "broker": it is unquestionably alive and it is not + // listening on the endpoint, so the readiness probe must fail. A broker that is + // mid-turn behaves the same way — its event loop is busy and it misses the probe + // window. Failing the probe must not be treated as "the broker is gone". + // Keep the broker's own session dir separate from the workspace: the buggy path + // unlinks sessionDir, and we still need the workspace to exist afterwards. + const sessionDir = path.join(workspace, "broker-session"); + fs.mkdirSync(sessionDir, { recursive: true }); + const session = { + endpoint: "unix:/tmp/codex-plugin-test-not-listening/broker.sock", + pidFile: path.join(sessionDir, "broker.pid"), + logFile: path.join(sessionDir, "broker.log"), + sessionDir, + pid: process.pid + }; + fs.writeFileSync(session.pidFile, String(process.pid)); + fs.writeFileSync(session.logFile, ""); + saveBrokerSession(workspace, session); + + let killed = false; + const result = await ensureBrokerSession(workspace, { + killProcess: () => { + killed = true; + } + }); + + assert.equal(killed, false, "a live broker must never be killed after a failed probe"); + assert.equal(result?.pid, process.pid, "the live broker session must be reused"); + assert.equal(loadBrokerSession(workspace)?.pid, process.pid, "its state must be kept"); + assert.equal(fs.existsSync(session.logFile), true, "its log must not be unlinked"); +}); + +test("ensureBrokerSession serializes concurrent callers so only one broker is created", async () => { + const workspace = makeTempDir(); + const created = []; + let counter = 0; + + // Both callers start with no persisted session. Without the lock they each spawn a + // broker and the last writer wins, orphaning the other broker and its app-server. + const options = { + createBrokerEndpoint: (sessionDir) => `unix:${sessionDir}/broker.sock`, + scriptPath: path.join(workspace, "fake-broker.mjs"), + timeoutMs: 1000 + }; + fs.writeFileSync( + options.scriptPath, + [ + 'import net from "node:net";', + 'const endpointIndex = process.argv.indexOf("--endpoint");', + 'const target = process.argv[endpointIndex + 1].replace(/^unix:/, "");', + "net.createServer(() => {}).listen(target);", + "setTimeout(() => {}, 60000);" + ].join("\n") + ); + + const spy = { ...options, onSpawn: () => created.push(++counter) }; + const [a, b] = await Promise.all([ + ensureBrokerSession(workspace, spy), + ensureBrokerSession(workspace, spy) + ]); + + assert.ok(a, "first caller must get a session"); + assert.ok(b, "second caller must get a session"); + assert.equal(a.endpoint, b.endpoint, "concurrent callers must share one broker"); + + for (const session of [a, b]) { + try { + if (session?.pid) process.kill(session.pid); + } catch { + // Already gone. + } + } +});