A previously hung POST with a given JSON-RPC id (e.g. id: 1, which Cowork reuses across calls) used to hold inFlight forever, so any subsequent call with the same id would block on Promise.all(previous) indefinitely. Mostly moot now that the engine race fix prevents the root hang, but keep this as defense in depth: cap the wait at 60s and log if we hit the cap, then proceed with the new request. Constructor accepts an override so tests can drive the timeout to 100ms instead of waiting wall-clock 60s in CI.
Unified · Split
| @@ -0,0 +1,107 @@ | |||
| 1 | import { describe, expect, it } from "vitest"; | ||
| 2 | |||
| 3 | import { JsonRpcRequestIdQueue, PREVIOUS_REQUEST_TIMEOUT_MS } from "./mcp-worker-transport"; | ||
| 4 | |||
| 5 | const jsonRpcRequest = (body: unknown): Request => | ||
| 6 | new Request("https://example.invalid/mcp", { | ||
| 7 | method: "POST", | ||
| 8 | headers: { "content-type": "application/json" }, | ||
| 9 | body: JSON.stringify(body), | ||
| 10 | }); | ||
| 11 | |||
| 12 | describe("JsonRpcRequestIdQueue", () => { | ||
| 13 | it("serialises requests with the same id", async () => { | ||
| 14 | const queue = new JsonRpcRequestIdQueue(); | ||
| 15 | const order: string[] = []; | ||
| 16 | |||
| 17 | let releaseFirst!: () => void; | ||
| 18 | const firstStarted = new Promise<void>((resolve) => { | ||
| 19 | const firstRunning = new Promise<void>((release) => { | ||
| 20 | releaseFirst = release; | ||
| 21 | }); | ||
| 22 | queue.run(jsonRpcRequest({ jsonrpc: "2.0", id: 1, method: "tools/call" }), async () => { | ||
| 23 | order.push("first:start"); | ||
| 24 | resolve(); | ||
| 25 | await firstRunning; | ||
| 26 | order.push("first:end"); | ||
| 27 | }); | ||
| 28 | }); | ||
| 29 | |||
| 30 | await firstStarted; | ||
| 31 | |||
| 32 | const secondDone = queue.run( | ||
| 33 | jsonRpcRequest({ jsonrpc: "2.0", id: 1, method: "tools/call" }), | ||
| 34 | async () => { | ||
| 35 | order.push("second"); | ||
| 36 | }, | ||
| 37 | ); | ||
| 38 | |||
| 39 | // Second must wait for first. | ||
| 40 | await new Promise((r) => setTimeout(r, 20)); | ||
| 41 | expect(order).toEqual(["first:start"]); | ||
| 42 | |||
| 43 | releaseFirst(); | ||
| 44 | await secondDone; | ||
| 45 | expect(order).toEqual(["first:start", "first:end", "second"]); | ||
| 46 | }); | ||
| 47 | |||
| 48 | it("does not block requests with different ids", async () => { | ||
| 49 | const queue = new JsonRpcRequestIdQueue(); | ||
| 50 | let release!: () => void; | ||
| 51 | const hung = new Promise<void>((resolve) => { | ||
| 52 | release = resolve; | ||
| 53 | }); | ||
| 54 | |||
| 55 | queue.run(jsonRpcRequest({ jsonrpc: "2.0", id: 1, method: "tools/call" }), () => hung); | ||
| 56 | |||
| 57 | const otherDone = await Promise.race([ | ||
| 58 | queue | ||
| 59 | .run(jsonRpcRequest({ jsonrpc: "2.0", id: 2, method: "tools/call" }), async () => "done") | ||
| 60 | .then((v) => ({ kind: "settled" as const, v })), | ||
| 61 | new Promise<{ kind: "blocked" }>((r) => | ||
| 62 | setTimeout(() => r({ kind: "blocked" }), 200), | ||
| 63 | ), | ||
| 64 | ]); | ||
| 65 | |||
| 66 | expect(otherDone.kind).toBe("settled"); | ||
| 67 | release(); | ||
| 68 | }); | ||
| 69 | |||
| 70 | it("regression: caps wait on a hung previous request and dispatches anyway", async () => { | ||
| 71 | // Override the timeout for fast CI — the production default is | ||
| 72 | // PREVIOUS_REQUEST_TIMEOUT_MS (60s) which we cap test-side to 100ms. | ||
| 73 | // Same behaviour, same code path; only the wall-clock budget changes. | ||
| 74 | const queue = new JsonRpcRequestIdQueue({ previousTimeoutMs: 100 }); | ||
| 75 | const order: string[] = []; | ||
| 76 | |||
| 77 | // Kick off a request and never release it — the poisoned-queue | ||
| 78 | // shape that used to cascade for the full upstream 180s timeout. | ||
| 79 | const firstStarted = new Promise<void>((resolve) => { | ||
| 80 | queue.run(jsonRpcRequest({ jsonrpc: "2.0", id: 1, method: "tools/call" }), async () => { | ||
| 81 | order.push("first:start"); | ||
| 82 | resolve(); | ||
| 83 | await new Promise(() => undefined); // hang forever | ||
| 84 | }); | ||
| 85 | }); | ||
| 86 | await firstStarted; | ||
| 87 | |||
| 88 | const result = await queue.run( | ||
| 89 | jsonRpcRequest({ jsonrpc: "2.0", id: 1, method: "tools/call" }), | ||
| 90 | async () => { | ||
| 91 | order.push("second"); | ||
| 92 | return "ok"; | ||
| 93 | }, | ||
| 94 | ); | ||
| 95 | |||
| 96 | expect(result).toBe("ok"); | ||
| 97 | expect(order).toEqual(["first:start", "second"]); | ||
| 98 | }); | ||
| 99 | |||
| 100 | it("exposes a sane production timeout", () => { | ||
| 101 | // Sanity guard: must stay below the 180s upstream timeout that | ||
| 102 | // Claude / Cowork enforce, but be long enough to outlast a normal | ||
| 103 | // dynamic-worker execution under load. | ||
| 104 | expect(PREVIOUS_REQUEST_TIMEOUT_MS).toBeGreaterThan(10_000); | ||
| 105 | expect(PREVIOUS_REQUEST_TIMEOUT_MS).toBeLessThan(180_000); | ||
| 106 | }); | ||
| 107 | }); | ||
| @@ -79,8 +79,22 @@ const extractJsonRpcRequestIdKeys = async (request: Request): Promise<ReadonlyAr · 78 unmodified lines | |||
| 79 | } | 79 | } |
| 80 | }; | 80 | }; |
| 81 | 81 | ||
| 82 | class JsonRpcRequestIdQueue { | 82 | // Hard ceiling on how long a same-id JSON-RPC request will wait for an |
| 83 | // earlier in-flight one to finish. Stays well under the 180s upstream | ||
| 84 | // client timeout that Claude / Cowork enforce, so a poisoned queue slot | ||
| 85 | // can't block the next request long enough for the client to give up. | ||
| 86 | // If a previous request hasn't released within the budget, we proceed | ||
| 87 | // anyway — at worst the MCP SDK rejects the second reply for a duplicate | ||
| 88 | // id, which is recoverable; a perma-stuck queue is not. | ||
| 89 | export const PREVIOUS_REQUEST_TIMEOUT_MS = 60_000; | ||
| 90 | |||
| 91 | export class JsonRpcRequestIdQueue { | ||
| 83 | private readonly inFlight = new Map<string, Promise<void>>(); | 92 | private readonly inFlight = new Map<string, Promise<void>>(); |
| 93 | private readonly previousTimeoutMs: number; | ||
| 94 | |||
| 95 | constructor(options: { readonly previousTimeoutMs?: number } = {}) { | ||
| 96 | this.previousTimeoutMs = options.previousTimeoutMs ?? PREVIOUS_REQUEST_TIMEOUT_MS; | ||
| 97 | } | ||
| 84 | 98 | ||
| 85 | async run<A>(request: Request, run: () => Promise<A>): Promise<A> { | 99 | async run<A>(request: Request, run: () => Promise<A>): Promise<A> { |
| 86 | const ids = [...new Set(await extractJsonRpcRequestIdKeys(request))]; | 100 | const ids = [...new Set(await extractJsonRpcRequestIdKeys(request))]; |
| @@ -96,7 +110,18 @@ class JsonRpcRequestIdQueue { · 9 unmodified lines | |||
| 96 | } | 110 | } |
| 97 | 111 | ||
| 98 | try { | 112 | try { |
| 99 | await Promise.all(previous.map((p) => p.catch(() => undefined))); | 113 | if (previous.length > 0) { |
| 114 | const settled = Promise.all(previous.map((p) => p.catch(() => undefined))); | ||
| 115 | const timeout = new Promise<"timeout">((resolve) => | ||
| 116 | setTimeout(() => resolve("timeout"), this.previousTimeoutMs), | ||
| 117 | ); | ||
| 118 | const outcome = await Promise.race([settled.then(() => "settled" as const), timeout]); | ||
| 119 | if (outcome === "timeout") { | ||
| 120 | console.warn( | ||
| 121 | `[mcp-worker-transport] previous in-flight request for ids=${JSON.stringify(ids)} did not release within ${this.previousTimeoutMs}ms; proceeding anyway`, | ||
| 122 | ); | ||
| 123 | } | ||
| 124 | } | ||
| 100 | return await run(); | 125 | return await run(); |
| 101 | } finally { | 126 | } finally { |
| 102 | for (const id of ids) { | 127 | for (const id of ids) { |