diff --git a/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs index 7c116b1dbd..bd7b4ac8ad 100644 --- a/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs @@ -13,6 +13,11 @@ // id; a test sets a wrong `sid` to prove the host drops a mismatched // notification and counts a protocol error. // - `exitCode`: when set, the fixture emits an exit notification after the data. +// - `dataAfterExit`: an array of `{ chunk, sid? }`, same shape as `data`. The +// fixture emits these as data notifications after the exit notification, so a +// test can script a batch where a data frame arrives after the exit in the +// read order — proving the host still bounds these frames on their own terms +// and never lets the exit crowd a data frame out of the pre-open hold. // - `echoInput`: when true, the fixture echoes each `duplexChannelWrite` back as // one data notification for the bound session. // - `closeMode`: "ack" | "bad-ack" | "no-ack" (default "ack"). It controls the @@ -47,6 +52,14 @@ function scriptedFrameLines(directive, workerSessionId) { params: { workerSessionId, exitCode: directive.exitCode }, })}\n`; } + const dataAfterExit = Array.isArray(directive.dataAfterExit) ? directive.dataAfterExit : []; + for (const entry of dataAfterExit) { + lines += `${JSON.stringify({ + jsonrpc: "2.0", + method: "duplexChannel.data", + params: { workerSessionId: entry.sid ?? workerSessionId, chunk: entry.chunk }, + })}\n`; + } return lines; } diff --git a/server/src/__tests__/plugin-worker-manager-duplex.test.ts b/server/src/__tests__/plugin-worker-manager-duplex.test.ts index b6a5c86a37..947315de5a 100644 --- a/server/src/__tests__/plugin-worker-manager-duplex.test.ts +++ b/server/src/__tests__/plugin-worker-manager-duplex.test.ts @@ -512,6 +512,107 @@ describe("plugin worker manager duplex channel route", () => { } }); + it("ends the route when batched pre-bind frames pass the frame count bound", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedFrames: 2 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + // The worker batches the three data frames with the open reply, so all + // three frames arrive before the route binds. No listener attaches, so + // the replay buffers them. The third frame passes the frame-count bound + // and the route ends. The pre-open hold must not drop the third frame + // before the buffered bound can end the route. + batchWithOpenReply: true, + data: [{ chunk: "a" }, { chunk: "b" }, { chunk: "c" }], + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route on the frame count bound even when a caller raises that bound well past the module default", async () => { + // Regression test: the pre-open hold ceiling must track + // maxDuplexChannelPreBindFrames, not a fixed value. A fixed ceiling at or + // below this bound would drop the 13th frame in the hold before the replay + // ever applies the buffered bound to it, and the route would never end. + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedFrames: 12 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + batchWithOpenReply: true, + data: Array.from({ length: 13 }, (_, i) => ({ chunk: String(i) })), + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route on the frame count bound even when an exit notification arrives before the overflow data frame", async () => { + // Regression test: an exit notification must never share the pre-open hold's + // capacity with data frames. The worker here batches exactly + // maxPreBindBufferedFrames valid data frames (no violation), then its exit, + // then one more data frame that should trip the buffered-frame bound. If the + // exit consumed a hold slot, that last data frame would be dropped by the + // hold before the replay ever applies the buffered bound to it, and the + // route would end normally on the exit (exitCode: 0) instead of on the + // bound (exitCode: null). + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedFrames: 3 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + batchWithOpenReply: true, + data: [{ chunk: "a" }, { chunk: "b" }, { chunk: "c" }], + exitCode: 0, + dataAfterExit: [{ chunk: "overflow" }], + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("delivers a batched pre-bind chunk that a later listener drains before the byte cap ends the route", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxTotalDataBytes: 4 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + // The worker batches the two data frames with the open reply, so both + // frames arrive before the route binds and before a listener attaches. + // "€" is three bytes in UTF-8. The first chunk (3 bytes ≤ 4) buffers. + // The second chunk brings the total to 6 bytes (> 4), so the route ends. + // The route end must not discard the buffered first chunk. The listener + // attaches after the open resolves and drains the first chunk. + batchWithOpenReply: true, + workerSessionId: "ws-A", + data: [{ chunk: "€" }, { chunk: "€" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + expect(chunks).toEqual(["€"]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + // ------------------------------------------------------------------------- // Authoritative closure and worker retirement. // ------------------------------------------------------------------------- diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index dfbe2a444a..96f8c67a39 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -175,6 +175,23 @@ const MAX_DUPLEX_CHANNEL_PRE_BIND_CHARS = 8 * 1024 * 1024; * channel route before a data listener attaches. */ const MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES = 10_000; +/** + * The margin the pre-open hold ceiling keeps above the pre-bind buffered-frame + * bound (`maxDuplexChannelPreBindFrames`, see below). A worker can batch frames + * with the open reply, so the host reads them before it knows the bound worker + * session id and before it can apply the per-frame bounds. The host holds these + * frames and replays them after the bind. The hold ceiling bounds that hold, so + * a worker that floods frames before it replies to the open cannot make the + * host hold an unbounded number of frames. + * + * The hold ceiling is derived from the buffered bound, not a fixed constant: a + * fixed ceiling equal to or below a caller-configured (or even the default) + * buffered bound would let the hold drop the frame that should instead trip the + * buffered bound during replay, so the route would never end. One frame of + * margin is enough — it lets the frame that exceeds the buffered bound reach + * the hold, so the replay's buffered-bound check, not the hold, ends the route. + */ +const DUPLEX_CHANNEL_PRE_OPEN_HOLD_MARGIN_FRAMES = 1; /** * The default maximum number of in-flight host→worker requests for one duplex * channel route. A worker that never replies cannot make the host hold an @@ -811,6 +828,11 @@ export function createPluginWorkerHandle( const maxDuplexChannelPreBindFrames = options.duplexChannelLimits?.maxPreBindBufferedFrames ?? MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES; + // Always strictly above maxDuplexChannelPreBindFrames, including when a + // caller configures that bound at or above the module default. See + // DUPLEX_CHANNEL_PRE_OPEN_HOLD_MARGIN_FRAMES for why the margin must hold. + const maxDuplexChannelPreOpenHoldFrames = + maxDuplexChannelPreBindFrames + DUPLEX_CHANNEL_PRE_OPEN_HOLD_MARGIN_FRAMES; const maxDuplexChannelPendingRequests = options.duplexChannelLimits?.maxPendingRequests ?? MAX_DUPLEX_CHANNEL_PENDING_REQUESTS; @@ -1483,13 +1505,26 @@ export function createPluginWorkerHandle( listener: ((chunk: string) => void) | null; buffered: string[]; bufferedChars: number; - // Raw data and exit notifications that arrive before the route binds. The - // host reads the worker stdout line by line. The open reply and a data or - // exit notification can arrive in one read batch, so the host dispatches the - // notification before the deferred open-reply continuation flips the state to - // `open`. The host holds these frames here and replays them in order right - // after it binds the route, so a batched frame is never lost. + // Raw data notifications that arrive before the route binds. The host reads + // the worker stdout line by line. The open reply and a data notification can + // arrive in one read batch, so the host dispatches the notification before + // the deferred open-reply continuation flips the state to `open`. The host + // holds these frames here and replays them in order right after it binds the + // route, so a batched frame is never lost. Bounded by the pre-open hold + // ceiling — see `bufferPreOpenDuplexChannelNotification`. preOpen: JsonRpcNotification[]; + // The most recent exit notification that arrived before the route binds, or + // null. A single slot, not an array: only the last exit a worker sends before + // the bind is ever meaningful, so the host overwrites this on every pre-open + // exit instead of holding each one. That keeps a worker that batches many + // exit notifications before it replies to the open from growing this past one + // entry — the pre-open hold ceiling bounds only `preOpen`, so an exit that + // shared that array with data frames could otherwise consume a data frame's + // hold slot and let a data frame that should trip the buffered-frame bound + // get dropped by the hold instead. Replayed after `preOpen` drains, so data + // still delivers before the exit resolves the wait, matching a real worker's + // order. + preOpenExit: JsonRpcNotification | null; pendingRequests: number; protocolErrors: number; totalDataBytes: number; @@ -1536,9 +1571,14 @@ export function createPluginWorkerHandle( route.terminalized = true; route.state = "closed"; route.listener = null; - route.buffered = []; - route.bufferedChars = 0; + // Keep the buffered chunks that the host accepted before the route ended, so + // a listener that attaches after the end still drains them. A frame can end + // the route during the pre-open replay, before a listener attaches, and the + // buffered chunks the host accepted before that frame are valid data the + // listener must still receive. The buffered bytes stay bounded by the + // pre-bind buffered bound, and `onData` clears them once it drains them. route.preOpen = []; + route.preOpenExit = null; clearDuplexChannelLifetimeTimer(route); // A terminalized route reports a null exit code, which the caller treats as a // failure. @@ -1567,37 +1607,62 @@ export function createPluginWorkerHandle( } // Hold one data or exit notification that arrives before the route binds. The - // host replays the held frames in order after it binds the route. Bound the - // hold by the pre-bind frame count, so a worker that floods frames before it - // replies to the open cannot make the host hold an unbounded number of frames. - // Count one protocol error for each frame past the bound. + // host replays the held notifications after it binds the route: `preOpen` + // drains first, in order, then the held exit (if any) resolves the wait last — + // see `drainPreOpenDuplexChannelNotifications`. + // + // A data notification goes on `preOpen`, bounded by a pre-open ceiling derived + // from the pre-bind buffered-frame bound, so a worker that floods data frames + // before it replies to the open cannot make the host hold an unbounded number + // of them. Count one protocol error for each data frame past the ceiling. This + // ceiling is separate from the pre-bind buffered-frame bound, but it tracks it: + // the replay after the bind applies the buffered bound to each held frame, so + // the buffered bound ends the route when a caller lowers (or raises) it. The + // hold ceiling stays above the buffered bound by construction + // (maxDuplexChannelPreOpenHoldFrames = maxDuplexChannelPreBindFrames + margin), + // or it would drop a frame before the buffered bound can end the route. + // + // An exit notification never touches `preOpen`. It overwrites the single + // `preOpenExit` slot instead, so a worker that batches an exit among enough + // data frames to fill the hold cannot consume a data frame's hold slot: the + // ceiling above bounds `preOpen` alone, so it stays exactly the margin above + // the buffered bound regardless of how many exit notifications arrive pre-open. function bufferPreOpenDuplexChannelNotification( route: DuplexChannelRoute, notification: JsonRpcNotification, ): void { - if (route.preOpen.length >= maxDuplexChannelPreBindFrames) { + if (notification.method === DUPLEX_CHANNEL_EXIT_NOTIFICATION) { + route.preOpenExit = notification; + return; + } + if (route.preOpen.length >= maxDuplexChannelPreOpenHoldFrames) { recordDuplexChannelProtocolError(route); return; } route.preOpen.push(notification); } - // Replay the held pre-open frames in order right after the route binds. The - // route is `open` now, so each frame passes through the normal per-frame bounds - // and the session-identifier match. A frame that ends the route terminalizes - // it, and every later frame in the replay is a no-op, because the routing - // functions drop a frame when the route is not `open`. + // Replay the held pre-open notifications right after the route binds. The + // route is `open` now, so each notification passes through the normal + // per-frame bounds and the session-identifier match. Drain the held data + // frames first, in order, then replay the held exit (if any) last, so data + // still delivers before the exit resolves the wait, matching a real worker's + // order. A frame that ends the route terminalizes it, and every later + // notification in the replay is a no-op, because the routing functions drop a + // notification when the route is not `open`. function drainPreOpenDuplexChannelNotifications(route: DuplexChannelRoute): void { - if (route.preOpen.length === 0) return; - const pending = route.preOpen; - route.preOpen = []; - for (const notification of pending) { - if (notification.method === DUPLEX_CHANNEL_DATA_NOTIFICATION) { + if (route.preOpen.length > 0) { + const pending = route.preOpen; + route.preOpen = []; + for (const notification of pending) { routeDuplexChannelData(notification); - } else if (notification.method === DUPLEX_CHANNEL_EXIT_NOTIFICATION) { - routeDuplexChannelExit(notification); } } + if (route.preOpenExit) { + const exitNotification = route.preOpenExit; + route.preOpenExit = null; + routeDuplexChannelExit(exitNotification); + } } // Deliver one duplex channel chunk to the bound listener in isolation. A @@ -1713,6 +1778,7 @@ export function createPluginWorkerHandle( route.buffered = []; route.bufferedChars = 0; route.preOpen = []; + route.preOpenExit = null; clearDuplexChannelLifetimeTimer(route); settleRouteWait(route, { exitCode: null }); } @@ -1742,6 +1808,7 @@ export function createPluginWorkerHandle( buffered: [], bufferedChars: 0, preOpen: [], + preOpenExit: null, pendingRequests: 0, protocolErrors: 0, totalDataBytes: 0,