fix(server): end the duplex route on the pre-bind bounds when frames arrive before the bind (#11860)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The host and a plugin worker talk over a duplex channel route with bounds on buffered frames and total bytes > - A worker can batch its data and exit frames with the open reply, so those frames arrive before the route binds and before a listener attaches > - Two of the route bounds did not hold on that pre-bind path: a shared limit let the pre-open hold swallow an over-limit frame before the buffered-frame bound could end the route, and the route end discarded chunks a later listener still needed > - This pull request gives the pre-open hold its own ceiling above the buffered bound, and keeps the buffered chunks across a route end > - The benefit is a duplex route that enforces its bounds and preserves valid data, even when a worker batches frames ahead of the bind ## Linked Issues or Issue Description No existing GitHub issue covers this. Filing it directly here, following the bug report template. **What happened?** Two duplex channel route bounds in `server/src/services/plugin-worker-manager.ts` did not hold when the data and exit frames arrived in the open-reply read batch, before the route bound: - The pre-open hold and the pre-bind buffered-frame bound shared one limit. When a caller lowered the buffered bound, the hold dropped the overflow frame as a protocol error before the buffered bound could end the route, so the route never ended. - The route end discarded the buffered chunks. A frame can end the route during the replay, before a listener attaches, and the chunks the host accepted before that frame are valid data. **Expected behavior** The pre-open hold uses its own ceiling, above the buffered bound, so the replay after the bind lets the buffered bound end the route. A route end keeps the buffered chunks so a listener that attaches after the end still drains them. **Steps to reproduce** 1. Open a duplex channel where the worker batches several data frames with the open reply. 2. Lower `maxPreBindBufferedFrames` below the batch size. 3. Observe the route fails to end on the buffered-frame bound, or a listener that attaches after an end-during-replay never receives the chunks buffered before that end. **Paperclip version or commit** `933749e01f74e82ce5d315c071be534d04e01158` **Deployment mode** Local dev (`pnpm dev`) and server unit tests. **Agent adapter(s) involved** None — this is host/plugin-worker transport infrastructure, not adapter-specific. **Database mode** Not database-related. **Access context** Any board or agent path that runs a plugin worker over a duplex channel route. ## What Changed - Give the pre-open frame hold its own ceiling (`MAX_DUPLEX_CHANNEL_PRE_OPEN_HOLD_FRAMES`), separate from the pre-bind buffered-frame bound, so lowering the buffered bound still ends the route instead of being pre-empted by the hold. - Keep the buffered chunks on a route end instead of discarding them, so a listener that attaches after an end-during-replay still drains the data the host already accepted. - Add two regression tests that batch frames with the open reply, so both bounds run through the pre-bind path deterministically. ## Verification - `cd server && npx vitest run src/__tests__/plugin-worker-manager-duplex.test.ts` — 24/24 tests pass, including the two new regression cases. ## Risks Low risk. This only changes bound bookkeeping on an internal transport path (frame hold ceiling and end-time buffer retention); it does not change the wire protocol or any public API. The new ceiling is a constant above the existing buffered bound, so pre-open holds are still capped. ## Model Used Claude, Sonnet 5 (claude-sonnet-5); assisted with repository-grounded diff review and drafted this PR description from the commit and code history. No functional code in this PR was authored by Claude — the fix itself is Priya Raman's, preserved with original authorship intact. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
9af1e75629
commit
33eb68b3ae
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
// -------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Reference in New Issue