fix(jeb): make upstream request deadlines configurable and preserve cancellation

This commit is contained in:
morluto committed 2026-10-15 05:05:18 +08:00
1 parent c831c23633
commit 097a4bce6f
9 files changed
+467 -45

No files matched your search

+9
View File
@@ -22,6 +22,15 @@ inspection and decompilation while refusing `open_jeb_project`. Catalog discover
uses one 60-second deadline across connection and pagination; a responsive server
cannot keep discovery running indefinitely by returning more pages.
Each upstream tool call has an absolute 60-second request deadline by default.
Set `REA_JEB_MCP_TIMEOUT_MS` to a positive decimal number of milliseconds to
select a different per-request budget, for example `REA_JEB_MCP_TIMEOUT_MS=300000`
for a five-minute decompilation. This budget starts after discovery and applies
to each individual `tools/call`; it does not extend the separate catalog deadline
or establish one end-to-end deadline across a whole CLI or MCP operation. A
request that reaches its budget returns a typed timeout, while caller
cancellation remains a cancellation.
Verified against JEB 5.48.0 (Linux x64, licensed build) serving `jeb-mcp-server` 1.3.0 over streamable HTTP.
## Operations
+32 -22
View File
@@ -1,8 +1,9 @@
import type { JebAnalysisPort } from "../application/jeb/JebAnalysisPort.js";
import { parseJebMcpTimeoutMs } from "../config/environment.js";
import { ConfigurationError } from "../domain/configurationErrors.js";
import { AnalysisInputError } from "../domain/analysisErrorCore.js";
import { err } from "../domain/result.js";
import type { JebMcpConnectionFactory } from "../jeb/JebMcpConnection.js";
import { parseJebMcpEndpoint } from "../jeb/JebMcpEndpoint.js";
import { JebProvider } from "../jeb/JebProvider.js";
/**
@@ -14,35 +15,44 @@ export const createJebAnalysisProvider = (
environment: Readonly<Record<string, string | undefined>> = process.env,
connectionFactory?: JebMcpConnectionFactory,
): JebAnalysisPort => {
const endpoint = parseJebMcpEndpoint(environment.REA_JEB_MCP_URL);
if (!endpoint.ok) {
const requestTimeout = parseJebMcpTimeoutMs(
environment.REA_JEB_MCP_TIMEOUT_MS,
);
if (!requestTimeout.ok) return unconfiguredJebProvider(requestTimeout.error);
try {
return new JebProvider({
environment,
...(requestTimeout.value === undefined
? {}
: { requestTimeoutMs: requestTimeout.value }),
...(connectionFactory === undefined ? {} : { connectionFactory }),
});
} catch (cause: unknown) {
if (!(cause instanceof AnalysisInputError)) throw cause;
const constraint =
typeof endpoint.error.cause === "string"
? endpoint.error.cause
: endpoint.error.message;
typeof cause.cause === "string" ? cause.cause : cause.message;
return unconfiguredJebProvider(
new ConfigurationError("Invalid REA environment configuration", {
cause: endpoint.error,
cause,
settings: [{ setting: "REA_JEB_MCP_URL", constraint }],
}),
);
}
return new JebProvider({
environment,
...(connectionFactory === undefined ? {} : { connectionFactory }),
});
};
const unconfiguredJebProvider = (
error: ConfigurationError,
): JebAnalysisPort => ({
inspectAvailability: () =>
Promise.resolve({
status: "unavailable",
code: "not_configured",
reason: `REA_JEB_MCP_URL is invalid: ${error.settings[0]?.constraint ?? error.message}`,
diagnostics: { setting: "REA_JEB_MCP_URL" },
}),
close: () => Promise.resolve(),
execute: () => Promise.resolve(err(error)),
});
): JebAnalysisPort => {
const setting = error.settings[0]?.setting ?? "REA_JEB_MCP_URL";
return {
inspectAvailability: () =>
Promise.resolve({
status: "unavailable",
code: "not_configured",
reason: `${setting} is invalid: ${error.settings[0]?.constraint ?? error.message}`,
diagnostics: { setting },
}),
close: () => Promise.resolve(),
execute: () => Promise.resolve(err(error)),
};
};
+38
View File
@@ -4,6 +4,10 @@ import {
defaultHopperLauncherPath,
parseConfig,
} from "./config/parseConfig.js";
import {
parseEnvironmentValues,
parseJebMcpTimeoutMs,
} from "./config/environment.js";
import { projectAnalysisError } from "./domain/analysisErrorProjection.js";
it("reports a configuration failure for a NUL in Hopper loader arguments", () => {
@@ -328,6 +332,40 @@ describe("configuration failure diagnostics", () => {
});
});
describe("JEB MCP timeout configuration", () => {
it("stays optional and validates the caller-selected environment value", () => {
expect(parseJebMcpTimeoutMs(undefined)).toEqual({
ok: true,
value: undefined,
});
expect(parseJebMcpTimeoutMs("900000")).toEqual({
ok: true,
value: 900_000,
});
expect(
parseEnvironmentValues({ REA_JEB_MCP_TIMEOUT_MS: "900000" }),
).toMatchObject({ ok: true, value: { REA_JEB_MCP_TIMEOUT_MS: 900_000 } });
});
it.each(["", "0", "-1", "1.5", "1e3", "invalid", "2147483648"])(
"rejects malformed timeout %j with its setting",
(value) => {
const result = parseJebMcpTimeoutMs(value);
if (result.ok) throw new Error("Expected invalid JEB MCP timeout");
expect(projectAnalysisError(result.error)).toMatchObject({
details: {
settings: [
{
setting: "REA_JEB_MCP_TIMEOUT_MS",
constraint: expect.any(String),
},
],
},
});
},
);
});
describe("default Linux Hopper launcher", () => {
const home = "/fixture/home";
const withExecutables =
+22
View File
@@ -30,6 +30,7 @@ const environmentSchema = z.object({
.min(1)
.refine(isAbsolute, "REA_IDA_MCP_CONFIG must be absolute")
.optional(),
REA_JEB_MCP_TIMEOUT_MS: providerStartupTimeoutSchema.optional(),
GHIDRA_INSTALL_DIR: z
.string()
.min(1)
@@ -77,6 +78,27 @@ const environmentSchema = z.object({
REA_REFERENCE_SECRET_PATTERNS_JSON: z.string().default("[]"),
});
/** Parse the JEB-only timeout at its isolated provider-composition boundary. */
export const parseJebMcpTimeoutMs = (
value: string | undefined,
): Result<number | undefined, ConfigurationError> => {
if (value === undefined) return ok(undefined);
const parsed = providerStartupTimeoutSchema.safeParse(value);
if (parsed.success) return ok(parsed.data);
return err(
new ConfigurationError("Invalid REA environment configuration", {
cause: parsed.error,
settings: [
{
setting: "REA_JEB_MCP_TIMEOUT_MS",
constraint:
parsed.error.issues[0]?.message ?? "must be a valid timeout",
},
],
}),
);
};
type Environment = Omit<
z.infer<typeof environmentSchema>,
"HOPPER_LOADER_ARGS_JSON" | "REA_REFERENCE_SECRET_PATTERNS_JSON"
+255 -1
View File
@@ -5,7 +5,171 @@ import {
decodeJebToolResult,
createStreamableHttpJebMcpConnection,
} from "./JebMcpConnection.js";
import { AnalysisProtocolError } from "../domain/analysisErrorCore.js";
import {
AnalysisCancelledError,
AnalysisProtocolError,
AnalysisTimeoutError,
} from "../domain/analysisErrorCore.js";
const jebTools = [
"get_client_information",
"open_project",
"list_units",
"decompile_code_item",
] as const;
const waitForRequestCount = async (
readCount: () => number,
expected: number,
method: string,
): Promise<void> => {
const deadline = performance.now() + 1_000;
while (readCount() < expected && performance.now() < deadline)
await new Promise((wake) => setTimeout(wake, 5));
if (readCount() < expected)
throw new Error(`Expected ${expected} ${method} requests`);
};
const startJebHttpFixture = (
responseDelayMs: number,
catalogRefreshDelayMs = 0,
) => {
let callCount = 0;
let catalogCount = 0;
const pendingTimers = new Set<NodeJS.Timeout>();
const server = createServer((request, response) => {
const chunks: Buffer[] = [];
request.on("data", (chunk: Buffer) => chunks.push(chunk));
request.on("end", () => {
if (chunks.length === 0) {
response.writeHead(202).end();
return;
}
const message: {
readonly id?: string | number;
readonly method: string;
readonly params?: Record<string, unknown>;
} = JSON.parse(Buffer.concat(chunks).toString("utf8"));
if (message.method === "notifications/initialized") {
response.writeHead(202).end();
return;
}
if (message.method === "notifications/cancelled") {
response.writeHead(202).end();
return;
}
if (message.method === "initialize") {
response.writeHead(200, { "Content-Type": "application/json" }).end(
JSON.stringify({
jsonrpc: "2.0",
id: message.id,
result: {
protocolVersion: message.params?.protocolVersion,
capabilities: { tools: {} },
serverInfo: { name: "jeb-timeout-fixture", version: "1" },
},
}),
);
return;
}
if (message.method === "tools/list") {
catalogCount++;
let responseClosed = false;
response.on("close", () => {
if (!response.writableEnded) responseClosed = true;
});
const sendCatalog = () => {
if (responseClosed) return;
response.writeHead(200, { "Content-Type": "application/json" }).end(
JSON.stringify({
jsonrpc: "2.0",
id: message.id,
result: {
tools: jebTools.map((name) => ({
name,
inputSchema: { type: "object", properties: {} },
})),
},
}),
);
};
if (catalogCount > 1 && catalogRefreshDelayMs > 0) {
const timer = setTimeout(() => {
pendingTimers.delete(timer);
sendCatalog();
}, catalogRefreshDelayMs);
pendingTimers.add(timer);
} else {
sendCatalog();
}
return;
}
if (message.method !== "tools/call") {
response.writeHead(202).end();
return;
}
callCount++;
let responseClosed = false;
response.on("close", () => {
if (!response.writableEnded) {
responseClosed = true;
}
});
const timer = setTimeout(() => {
pendingTimers.delete(timer);
if (responseClosed) return;
response.writeHead(200, { "Content-Type": "application/json" }).end(
JSON.stringify({
jsonrpc: "2.0",
id: message.id,
result: {
structuredContent: { success: true, text: "finished" },
content: [{ type: "text", text: '{"success":true}' }],
},
}),
);
}, responseDelayMs);
pendingTimers.add(timer);
});
});
return new Promise<{
readonly endpoint: URL;
readonly callCount: () => number;
readonly waitForCall: (count?: number) => Promise<void>;
readonly waitForCatalog: (count?: number) => Promise<void>;
readonly close: () => Promise<void>;
}>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", () => {
const address = server.address();
if (address === null || typeof address === "string") {
reject(new Error("Expected TCP listener"));
return;
}
server.removeListener("error", reject);
resolve({
endpoint: new URL(`http://127.0.0.1:${address.port}/mcp`),
callCount: () => callCount,
waitForCall: (count = 1) =>
waitForRequestCount(() => callCount, count, "tools/call"),
waitForCatalog: (count = 1) =>
waitForRequestCount(() => catalogCount, count, "tools/list"),
close: async () => {
for (const timer of pendingTimers) clearTimeout(timer);
pendingTimers.clear();
const closed = new Promise<void>((done, fail) =>
server.close((cause) =>
cause === undefined ? done() : fail(cause),
),
);
server.closeAllConnections();
await closed;
},
});
});
});
};
describe("decodeJebToolResult", () => {
it("passes refusal envelopes through so the provider preserves the engine reason", () => {
@@ -118,3 +282,93 @@ it("refuses HTTP redirects before sending requests to another endpoint", async (
);
}
});
it("applies the selected timeout to real MCP tool calls", async () => {
const fixture = await startJebHttpFixture(180);
const short = createStreamableHttpJebMcpConnection(fixture.endpoint, {
requestTimeoutMs: 50,
});
const extended = createStreamableHttpJebMcpConnection(fixture.endpoint, {
requestTimeoutMs: 500,
});
try {
await short.connect();
await expect(short.call("decompile_code_item", {})).rejects.toBeInstanceOf(
AnalysisTimeoutError,
);
expect(fixture.callCount()).toBe(1);
await extended.connect();
await expect(
extended.call("decompile_code_item", {}),
).resolves.toMatchObject({ success: true, text: "finished" });
expect(fixture.callCount()).toBe(2);
} finally {
try {
await Promise.all([short.close(), extended.close()]);
} finally {
await fixture.close();
}
}
});
it("keeps caller abort and connection close as cancellations for active HTTP calls", async () => {
const fixture = await startJebHttpFixture(1_000);
const connection = createStreamableHttpJebMcpConnection(fixture.endpoint, {
requestTimeoutMs: 2_000,
});
try {
await connection.connect();
const alreadyAborted = new AbortController();
alreadyAborted.abort();
await expect(
connection.call("decompile_code_item", {}, alreadyAborted.signal),
).rejects.toBeInstanceOf(AnalysisCancelledError);
expect(fixture.callCount()).toBe(0);
const caller = new AbortController();
const callerCall = connection.call(
"decompile_code_item",
{},
caller.signal,
);
await fixture.waitForCall(1);
caller.abort();
await expect(callerCall).rejects.toBeInstanceOf(AnalysisCancelledError);
const closingCall = connection.call("decompile_code_item", {});
await fixture.waitForCall(2);
await connection.close();
await expect(closingCall).rejects.toBeInstanceOf(AnalysisCancelledError);
} finally {
try {
await connection.close();
} finally {
await fixture.close();
}
}
});
it("keeps caller abort and connection close as cancellations during catalog refresh", async () => {
const fixture = await startJebHttpFixture(0, 1_000);
const connection = createStreamableHttpJebMcpConnection(fixture.endpoint);
try {
await connection.connect();
const caller = new AbortController();
const callerRefresh = connection.listTools(caller.signal);
await fixture.waitForCatalog(2);
caller.abort();
await expect(callerRefresh).rejects.toBeInstanceOf(AnalysisCancelledError);
const closingRefresh = connection.listTools();
await fixture.waitForCatalog(3);
await connection.close();
await expect(closingRefresh).rejects.toBeInstanceOf(AnalysisCancelledError);
} finally {
try {
await connection.close();
} finally {
await fixture.close();
}
}
});
+64 -15
View File
@@ -1,6 +1,8 @@
import {
Client,
DEFAULT_REQUEST_TIMEOUT_MSEC,
SdkError,
SdkErrorCode,
StreamableHTTPClientTransport,
} from "@modelcontextprotocol/client";
import { z } from "zod";
@@ -82,16 +84,27 @@ const isJebFailureEnvelope = (value: JsonValue): boolean =>
!Array.isArray(value) &&
value.success === false;
export interface JebMcpConnectionTimeouts {
/** Absolute catalog connection and pagination deadline. */
readonly discoveryTimeoutMs?: number;
/** Absolute timeout for each upstream tools/call request. */
readonly requestTimeoutMs?: number;
}
export type JebMcpConnectionFactory = (
endpoint: URL,
timeoutMs?: number,
timeouts?: JebMcpConnectionTimeouts,
) => JebMcpConnection;
/** Serve JEB requests from a running engine client over streamable HTTP. */
export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
endpoint,
timeoutMs = DEFAULT_REQUEST_TIMEOUT_MSEC,
timeouts,
) => {
const discoveryTimeoutMs =
timeouts?.discoveryTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MSEC;
const requestTimeoutMs =
timeouts?.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MSEC;
interface ConnectionState {
readonly client: Client;
readonly transport: StreamableHTTPClientTransport;
@@ -184,7 +197,7 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
state = owner;
const acquisition = (async () => {
const deadline = new ProviderStartupDeadline(
timeoutMs,
discoveryTimeoutMs,
owner.signal.signal,
);
try {
@@ -197,7 +210,9 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
} catch (cause: unknown) {
const failure =
deadline.interruption === "timeout"
? new AnalysisTimeoutError("jeb_mcp_catalog", timeoutMs, { cause })
? new AnalysisTimeoutError("jeb_mcp_catalog", discoveryTimeoutMs, {
cause,
})
: cause;
owner.signal.abort(failure);
try {
@@ -234,19 +249,30 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
// could detach this waiter before that owner has disposed a failed client.
const tools = await connect(signal);
if (!existing) return tools;
const deadline = new ProviderStartupDeadline(timeoutMs, signal);
const owner = state;
if (owner === undefined)
throw new AnalysisProtocolError("JEB MCP client failed to connect");
const discoverySignal =
signal === undefined
? owner.signal.signal
: AbortSignal.any([signal, owner.signal.signal]);
const deadline = new ProviderStartupDeadline(
discoveryTimeoutMs,
discoverySignal,
);
try {
const owner = state;
if (owner === undefined)
throw new AnalysisProtocolError("JEB MCP client failed to connect");
const refreshed = await readToolNames(owner.client, deadline.signal);
owner.tools = refreshed;
return refreshed;
} catch (cause: unknown) {
if (deadline.interruption === "timeout")
throw new AnalysisTimeoutError("jeb_mcp_catalog", timeoutMs, {
cause,
});
throw new AnalysisTimeoutError(
"jeb_mcp_catalog",
discoveryTimeoutMs,
{ cause },
);
if (deadline.interruption === "cancelled")
throw new AnalysisCancelledError("jeb_mcp_catalog", { cause });
throw cause;
} finally {
deadline.dispose();
@@ -257,11 +283,34 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
const owner = state;
if (owner === undefined || owner.tools === undefined)
throw new AnalysisProtocolError("JEB MCP client failed to connect");
const result = await owner.client.callTool(
{ name, arguments: args },
signal === undefined ? {} : { signal },
const requestSignal =
signal === undefined
? owner.signal.signal
: AbortSignal.any([signal, owner.signal.signal]);
const deadline = new ProviderStartupDeadline(
requestTimeoutMs,
requestSignal,
);
return decodeJebToolResult(result);
try {
const result = await owner.client.callTool(
{ name, arguments: args },
{ timeout: deadline.remainingMs(), signal: deadline.signal },
);
return decodeJebToolResult(result);
} catch (cause: unknown) {
if (deadline.interruption === "timeout")
throw new AnalysisTimeoutError(name, requestTimeoutMs, { cause });
if (requestSignal.aborted)
throw new AnalysisCancelledError(name, { cause });
if (
cause instanceof SdkError &&
cause.code === SdkErrorCode.RequestTimeout
)
throw new AnalysisTimeoutError(name, requestTimeoutMs, { cause });
throw cause;
} finally {
deadline.dispose();
}
},
async close() {
closed = true;
+35 -1
View File
@@ -1,7 +1,11 @@
import { describe, expect, it } from "vitest";
import { createJebAnalysisProvider } from "../composition/jeb.js";
import { JebProvider } from "./JebProvider.js";
import type { JebMcpConnection } from "./JebMcpConnection.js";
import type {
JebMcpConnection,
JebMcpConnectionTimeouts,
} from "./JebMcpConnection.js";
import {
AnalysisCapabilityUnavailableError,
AnalysisInputError,
@@ -42,6 +46,36 @@ const providerWith = (connection: JebMcpConnection) =>
});
describe("JebProvider live availability", () => {
it("passes its validated environment timeout to the production connection", async () => {
let configured: JebMcpConnectionTimeouts | undefined;
const provider = createJebAnalysisProvider(
{ REA_JEB_MCP_TIMEOUT_MS: "900000" },
(_endpoint, timeouts) => {
configured = timeouts;
return scriptedConnection({});
},
);
expect(configured).toEqual({ requestTimeoutMs: 900_000 });
await provider.close();
});
it("rejects an invalid JEB timeout before constructing its MCP connection", async () => {
let constructed = false;
const provider = createJebAnalysisProvider(
{ REA_JEB_MCP_TIMEOUT_MS: "0" },
() => {
constructed = true;
return scriptedConnection({});
},
);
expect(constructed).toBe(false);
await expect(provider.inspectAvailability()).resolves.toMatchObject({
status: "unavailable",
diagnostics: { setting: "REA_JEB_MCP_TIMEOUT_MS" },
});
await provider.close();
});
it("reports only the operations advertised by the live engine", async () => {
const connection = scriptedConnection({
get_client_information: {
+10 -4
View File
@@ -39,6 +39,8 @@ export const JEB_PROVIDER_IDENTITY: ProviderIdentity = {
export interface JebProviderOptions {
/** Caller environment carrying REA_JEB_MCP_URL. */
readonly environment?: Readonly<Record<string, string | undefined>>;
/** Per-tools/call budget selected by REA_JEB_MCP_TIMEOUT_MS. */
readonly requestTimeoutMs?: number;
/** Injectable MCP boundary; defaults to the streamable HTTP client. */
readonly connectionFactory?: JebMcpConnectionFactory;
}
@@ -111,10 +113,14 @@ export class JebProvider implements JebAnalysisPort {
const endpoint = parseJebMcpEndpoint(environment.REA_JEB_MCP_URL);
if (!endpoint.ok) throw endpoint.error;
this.#endpoint = endpoint.value;
this.#connection = (
options.connectionFactory ??
((url: URL) => createStreamableHttpJebMcpConnection(url))
)(this.#endpoint);
const createConnection =
options.connectionFactory ?? createStreamableHttpJebMcpConnection;
this.#connection =
options.requestTimeoutMs === undefined
? createConnection(this.#endpoint)
: createConnection(this.#endpoint, {
requestTimeoutMs: options.requestTimeoutMs,
});
}
async inspectAvailability(
@@ -76,7 +76,7 @@ it("reads every JEB catalog page and refreshes the connected engine catalog", as
throw new Error("Expected an ephemeral TCP listener");
const connection = createStreamableHttpJebMcpConnection(
new URL(`http://127.0.0.1:${address.port}/mcp`),
1_000,
{ discoveryTimeoutMs: 1_000 },
);
try {
expect(await connection.listTools()).toEqual([
@@ -109,7 +109,7 @@ it("reads every JEB catalog page and refreshes the connected engine catalog", as
// Shared initial acquisition also owns a finite deadline and can recover.
const acquiring = createStreamableHttpJebMcpConnection(
new URL(`http://127.0.0.1:${address.port}/mcp`),
1_000,
{ discoveryTimeoutMs: 1_000 },
);
try {
infinitePages = true;