Skip to content

Commit b057f64

Browse files
axisrowclaude
axisrow
and
claude
authored
fix(broker): bound broker handshake so a wedged broker can't hang the client (closes #29) (#30)
* fix(broker): bound broker handshake so a wedged broker can't hang the client (closes #29) When the shared broker is wedged (connect accepted but initialize never answered), BrokerCodexAppServerClient had no deadline on net.createConnection or request("initialize"), so a second /codex:* command hung forever — the broker-busy hang seen in upstream openai#509. - Bound connect (2s default) and initialize (5s default) with a new BROKER_HANDSHAKE_TIMEOUT_CODE ("EBROKERTIMEOUT"); both overridable via brokerConnectTimeoutMs / brokerInitializeTimeoutMs options. Timers are unref'd so they don't pin the event loop. - withAppServer now falls back to a direct app-server on EBROKERTIMEOUT in addition to BROKER_BUSY_RPC_CODE / ENOENT / ECONNREFUSED. - RED-first: spawnWedgedServer helper + broker-client-initialize-timeout test reproduces the hang (connect ok, no initialize answer → EBROKERTIMEOUT). Follow-up (not in this PR): SpawnedCodexAppServerClient.initialize has the same unbounded shape; a deeper fix would put requestTimeoutMs on AppServerClientBase.request() and collapse the withAppServer OR-chain. Co-Authored-By: Claude <[email protected]> * fix(broker): make the EBROKERTIMEOUT direct-fallback reachable (cycle-review critical) Codex adversarial review caught that the handshake timeout rejects inside CodexAppServerClient.connect() — before `client` is assigned in withAppServer — so the `client?.transport === "broker"` guard was always false for EBROKERTIMEOUT. The fallback clause was dead code: a wedged broker made every withAppServer-backed command throw after 5s instead of retrying directly. Tag initialization errors with the chosen transport in connect(), and let the fallback match on error.transport when client is still null. Adds an end-to-end test (wedged broker → withAppServer → direct fallback via fake-codex). Co-Authored-By: Claude <[email protected]> --------- Co-authored-by: axisrow <[email protected]> Co-authored-by: Claude <[email protected]>
1 parent e42a23b commit b057f64

4 files changed

Lines changed: 184 additions & 7 deletions

File tree

‎plugins/codex/scripts/lib/app-server.mjs‎

Lines changed: 57 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,14 @@ const PLUGIN_MANIFEST = JSON.parse(fs.readFileSync(PLUGIN_MANIFEST_URL, "utf8"))
2222
export const BROKER_ENDPOINT_ENV = "CODEX_COMPANION_APP_SERVER_ENDPOINT";
2323
export const BROKER_BUSY_RPC_CODE = -32001;
2424

25+
// Error code attached to timeouts during the broker handshake, so withAppServer
26+
// can recognize them and fall back to a direct (non-broker) app-server. A
27+
// wedged broker — connect accepted but initialize never answered — must not
28+
// hang a second /codex:* command forever (fork #29 / upstream openai#509).
29+
export const BROKER_HANDSHAKE_TIMEOUT_CODE = "EBROKERTIMEOUT";
30+
const BROKER_CONNECT_TIMEOUT_MS = 2000;
31+
const BROKER_INITIALIZE_TIMEOUT_MS = 5000;
32+
2533
/** @type {ClientInfo} */
2634
const DEFAULT_CLIENT_INFO = {
2735
title: "Codex Plugin",
@@ -283,29 +291,62 @@ class BrokerCodexAppServerClient extends AppServerClientBase {
283291
}
284292

285293
async initialize() {
294+
const handshakeError = (message) => {
295+
const error = new Error(message);
296+
error.code = BROKER_HANDSHAKE_TIMEOUT_CODE;
297+
return error;
298+
};
299+
300+
const connectTimeoutMs = this.options.brokerConnectTimeoutMs ?? BROKER_CONNECT_TIMEOUT_MS;
301+
const initializeTimeoutMs = this.options.brokerInitializeTimeoutMs ?? BROKER_INITIALIZE_TIMEOUT_MS;
302+
303+
// Bound the connect step: a wedged broker may accept the socket but never
304+
// signal connect — destroy and reject so the caller can fall back.
286305
await new Promise((resolve, reject) => {
287306
const target = parseBrokerEndpoint(this.endpoint);
288307
this.socket = net.createConnection({ path: target.path });
289308
this.socket.setEncoding("utf8");
290-
this.socket.on("connect", resolve);
309+
const connectTimer = setTimeout(() => {
310+
this.socket.destroy();
311+
reject(handshakeError("codex app-server broker connect timed out."));
312+
}, connectTimeoutMs).unref?.();
313+
this.socket.on("connect", () => {
314+
clearTimeout(connectTimer);
315+
resolve();
316+
});
291317
this.socket.on("data", (chunk) => {
292318
this.handleChunk(chunk);
293319
});
294320
this.socket.on("error", (error) => {
321+
clearTimeout(connectTimer);
295322
if (!this.exitResolved) {
296323
reject(error);
297324
}
298325
this.handleExit(error);
299326
});
300327
this.socket.on("close", () => {
328+
clearTimeout(connectTimer);
301329
this.handleExit(this.exitError);
302330
});
303331
});
304332

305-
await this.request("initialize", {
306-
clientInfo: this.options.clientInfo ?? DEFAULT_CLIENT_INFO,
307-
capabilities: this.options.capabilities ?? DEFAULT_CAPABILITIES
333+
// Bound the initialize request: a wedged broker may connect but never
334+
// answer. Race the request against a deadline; on timeout, force the
335+
// connection closed (handleExit rejects the pending request).
336+
let initializeTimer;
337+
const initializeDeadline = new Promise((_, reject) => {
338+
initializeTimer = setTimeout(() => {
339+
this.socket.destroy();
340+
reject(handshakeError("codex app-server initialize timed out."));
341+
}, initializeTimeoutMs).unref?.();
308342
});
343+
await Promise.race([
344+
this.request("initialize", {
345+
clientInfo: this.options.clientInfo ?? DEFAULT_CLIENT_INFO,
346+
capabilities: this.options.capabilities ?? DEFAULT_CAPABILITIES
347+
}),
348+
initializeDeadline
349+
]).finally(() => clearTimeout(initializeTimer));
309350
this.notify("initialized", {});
310351
}
311352

@@ -352,7 +393,18 @@ export class CodexAppServerClient {
352393
const client = brokerEndpoint
353394
? new BrokerCodexAppServerClient(cwd, { ...options, brokerEndpoint })
354395
: new SpawnedCodexAppServerClient(cwd, options);
355-
await client.initialize();
396+
try {
397+
await client.initialize();
398+
} catch (error) {
399+
// The handshake rejects before `connect()` returns, so callers like
400+
// withAppServer cannot read `client.transport` (client is never assigned
401+
// there). Tag the error with the transport so the caller can still tell
402+
// a broker-handshake failure (worth a direct fallback) from a spawned one.
403+
if (error && !error.transport) {
404+
error.transport = client.transport;
405+
}
406+
throw error;
407+
}
356408
return client;
357409
}
358410
}

‎plugins/codex/scripts/lib/codex.mjs‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ import os from "node:os";
4040
import path from "node:path";
4141

4242
import { readJsonFile } from "./fs.mjs";
43-
import { BROKER_BUSY_RPC_CODE, BROKER_ENDPOINT_ENV, CodexAppServerClient } from "./app-server.mjs";
43+
import { BROKER_BUSY_RPC_CODE, BROKER_ENDPOINT_ENV, BROKER_HANDSHAKE_TIMEOUT_CODE, CodexAppServerClient } from "./app-server.mjs";
4444
import { loadBrokerSession } from "./broker-lifecycle.mjs";
4545
import { binaryAvailable } from "./process.mjs";
4646
import { validateExplicitReasoningSelection, validateReasoningSelection } from "./model-catalog.mjs";
@@ -679,7 +679,7 @@ async function captureTurn(client, threadId, startRequest, options = {}) {
679679
}
680680
}
681681

682-
async function withAppServer(cwd, fn, clientOptions = {}) {
682+
export async function withAppServer(cwd, fn, clientOptions = {}) {
683683
let client = null;
684684
try {
685685
client = await CodexAppServerClient.connect(cwd, clientOptions);
@@ -688,8 +688,11 @@ async function withAppServer(cwd, fn, clientOptions = {}) {
688688
return result;
689689
} catch (error) {
690690
const brokerRequested = client?.transport === "broker" || Boolean(process.env[BROKER_ENDPOINT_ENV]);
691+
const brokerHandshakeFailed =
692+
error?.code === BROKER_HANDSHAKE_TIMEOUT_CODE && (client?.transport === "broker" || error?.transport === "broker");
691693
const shouldRetryDirect =
692694
(client?.transport === "broker" && error?.rpcCode === BROKER_BUSY_RPC_CODE) ||
695+
brokerHandshakeFailed ||
693696
(brokerRequested && (error?.code === "ENOENT" || error?.code === "ECONNREFUSED"));
694697

695698
if (client) {

‎tests/broker-client-initialize-timeout.test.mjs‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
import test from "node:test";
2+
import assert from "node:assert/strict";
3+
4+
import { CodexAppServerClient } from "../plugins/codex/scripts/lib/app-server.mjs";
5+
import { withAppServer } from "../plugins/codex/scripts/lib/codex.mjs";
6+
import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs";
7+
import { makeTempDir } from "./helpers.mjs";
8+
import { spawnWedgedServer } from "./broker-controller-helpers.mjs";
9+
10+
// Reproduces fork #29 / upstream openai#509: when the shared broker is wedged
11+
// (busy with a long-running request and never reaching the second socket), the
12+
// client must not hang forever on connect + initialize. This is the RED test —
13+
// today there is no deadline on net.createConnection or request("initialize").
14+
test("client does not hang on initialize when broker is wedged (#29/#509)", async () => {
15+
const wedged = await spawnWedgedServer();
16+
const cwd = makeTempDir();
17+
// Short handshake budgets so the test does not wait on the production
18+
// defaults (2s connect / 5s initialize). The wedged broker connects but
19+
// never answers initialize, so the initialize deadline must fire.
20+
const connectP = CodexAppServerClient.connect(cwd, {
21+
brokerEndpoint: `unix:${wedged.socketPath}`,
22+
brokerConnectTimeoutMs: 300,
23+
brokerInitializeTimeoutMs: 500
24+
});
25+
26+
let outcome = "HANG";
27+
try {
28+
const winner = await Promise.race([
29+
connectP.then(
30+
() => "connected",
31+
(error) => ({ errored: error.code ?? error.rpcCode ?? error.message })
32+
),
33+
new Promise((resolve) => setTimeout(() => resolve("HANG"), 2000))
34+
]);
35+
outcome = winner;
36+
} finally {
37+
wedged.cleanup();
38+
}
39+
40+
assert.notEqual(outcome, "HANG", "initialize hung on a wedged broker (#29/#509)");
41+
assert.ok(
42+
outcome && typeof outcome === "object" && outcome.errored === "EBROKERTIMEOUT",
43+
`expected EBROKERTIMEOUT from wedged broker, got ${JSON.stringify(outcome)}`
44+
);
45+
});
46+
47+
// Drives the full withAppServer fallback path: a wedged broker makes the
48+
// broker-client handshake time out, and withAppServer must fall back to a
49+
// direct (spawned) app-server instead of propagating the timeout. Codex review
50+
// of PR #30 caught that the EBROKERTIMEOUT fallback was unreachable because the
51+
// handshake rejects before `client` is assigned, so client.transport is null.
52+
test("withAppServer falls back to a direct app-server when the broker wedges (#29/#509)", async () => {
53+
const wedged = await spawnWedgedServer();
54+
const cwd = makeTempDir();
55+
const binDir = makeTempDir();
56+
installFakeCodex(binDir);
57+
const env = buildEnv(binDir);
58+
59+
let calls = 0;
60+
try {
61+
const result = await Promise.race([
62+
withAppServer(
63+
cwd,
64+
async (client) => {
65+
calls += 1;
66+
return { transport: client.transport };
67+
},
68+
{
69+
brokerEndpoint: `unix:${wedged.socketPath}`,
70+
brokerConnectTimeoutMs: 300,
71+
brokerInitializeTimeoutMs: 500,
72+
env
73+
}
74+
),
75+
new Promise((resolve) => setTimeout(() => resolve({ HANG: true }), 5000))
76+
]);
77+
78+
assert.ok(!result.HANG, "withAppServer hung instead of falling back to direct");
79+
assert.equal(calls, 1, "the operation must run exactly once via the direct fallback");
80+
assert.equal(result.transport, "direct", "the operation must run on the direct transport after fallback");
81+
} finally {
82+
wedged.cleanup();
83+
}
84+
});

‎tests/broker-controller-helpers.mjs‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import fs from "node:fs";
12
import net from "node:net";
23
import path from "node:path";
34

@@ -193,3 +194,40 @@ export async function spawnInProcessBroker({ appClient, idleTimeoutMs = 0, endpo
193194
sendRpc
194195
};
195196
}
197+
198+
/**
199+
* A real net.createServer that accepts connections but never answers any
200+
* request — simulates a wedged broker (busy with a long-running request and
201+
* never reaching the second socket's data). Drives the client-side connect +
202+
* initialize path with no server response, exposing any missing deadline.
203+
*/
204+
export async function spawnWedgedServer() {
205+
const sessionDir = makeTempDir("broker-wedged-");
206+
const socketPath = path.join(sessionDir, "broker.sock");
207+
const connections = new Set();
208+
const server = net.createServer((socket) => {
209+
socket.setEncoding("utf8");
210+
connections.add(socket);
211+
socket.on("close", () => connections.delete(socket));
212+
// Intentionally ignore "data": the broker is wedged and never responds.
213+
socket.on("error", () => {});
214+
});
215+
await new Promise((resolve) => server.listen(socketPath, resolve));
216+
return {
217+
socketPath,
218+
cleanup() {
219+
// Destroy accepted sockets so the wedged client's request rejects and
220+
// the process can exit. server.close() alone would wait for them.
221+
for (const socket of connections) {
222+
socket.destroy();
223+
}
224+
connections.clear();
225+
try {
226+
server.close();
227+
} catch {}
228+
try {
229+
fs.unlinkSync(socketPath);
230+
} catch {}
231+
}
232+
};
233+
}

0 commit comments

Comments
 (0)