From 25d771d6a889e4e344007869cad21568eaec2f6e Mon Sep 17 00:00:00 2001 From: Tabitha-2696 Date: Fri, 16 Oct 2026 01:20:39 -0400 Subject: [PATCH 1/3] fix(mcpl): held requests get a budget of 10,000, apart from notifications; past it a request is refused (#387) While a plane is paused (startup, a reconnect, an awareness drain, a quiesce), emit() holds what a server sends until the plane opens. Issue #122's cap bounded only responder-less events: every request was appended past it, on the theory that a server times out and reconnects. A server that pipelines requests without waiting never times out, and a held request keeps its params and its responder closure, so memory grew until the host ran out. Probed on main 91e70eda: 5,000 push/event requests of 10 KB sent while paused were all held, none answered, and the heap grew 73 MB. Requests and notifications are now bounded apart: - Requests: past MAX_HELD_REQUESTS (10,000) a new request is answered at once with -32000, "the host is busy: 10000 of this server's requests are already waiting; send this one again later", the code of the host's other transient refusals, and is not held. It is never dropped, since its server waits on it, and the connection is never recycled: held items survive a reconnect, so recycling would free almost nothing and would cost a healthy, busy server its link. The first refusal and every hundredth are logged, as the drops are. - Notifications: the existing cap of 1,000 (now MAX_HELD_NOTIFICATIONS) counts held notifications only. Before, it counted held requests too, so with 1,000 or more requests held, each new notification evicted an older one or, with none held, was dropped itself. Now the oldest held notification makes room only when 1,000 notifications are held. A shared 1,000 would have turned a long pause over a busy guild into lost messages for a healthy server, since discord-mcpl sends one request per Discord message and only logs a refused one. 10,000 still bounds the pipelining server #387 is about. The changelog names the budget and the refusal. heldRequests counts the held requests, so deciding whether either bound is reached reads the count, not the buffer. ready() and readyControlPlane() zero it as they take the snapshot, and a request held again during their flush (flushDepth) is counted without being bounded again, since it was admitted once. Every change to bufferedEvents keeps the count; #250, open, filters bufferedEvents at a transport boundary and would need to recount there. The request predicate is a static hasResponder, the name and body #250 gives it. Tests (test/mcpl-held-request-budget.test.ts), on the connection's real routing with both planes closed and a push/event listener: - 10,000 requests are held, the next two are refused -32000 with the busy text and one log line, and ready() delivers the 10,000 in order; - with 10,000 requests held, 1,001 notifications are held as 1,000, the oldest dropped, nothing refused, and ready() delivers n2..n1001; - 5,000 requests held again by readyControlPlane() (data plane still paused) count: 5,000 more fill the budget and the next is refused; after ready(), a new pause holds 10,000 with no refusal. test/mcpl-request-handler-errors.test.ts held a request form beside 1,100 notifications and counted it against the 1,000, so 101 were dropped. The request is counted apart now: 100 are dropped, and the flush holds 1,001. Co-Authored-By: Claude Opus 5.5 --- changelog.d/387-held-request-budget.fixed.md | 11 ++ src/mcpl/server-connection.ts | 78 +++++++++---- test/mcpl-held-request-budget.test.ts | 109 +++++++++++++++++++ test/mcpl-request-handler-errors.test.ts | 10 +- 4 files changed, 180 insertions(+), 28 deletions(-) create mode 100644 changelog.d/387-held-request-budget.fixed.md create mode 100644 test/mcpl-held-request-budget.test.ts diff --git a/changelog.d/387-held-request-budget.fixed.md b/changelog.d/387-held-request-budget.fixed.md new file mode 100644 index 000000000..80dd592c7 --- /dev/null +++ b/changelog.d/387-held-request-budget.fixed.md @@ -0,0 +1,11 @@ +- **Requests an MCPL server sends while its plane is paused are bounded** + (#387). During a startup, a reconnect, an awareness drain or a quiesce, + held requests had no bound, so a server that sends requests without + waiting for their answers grew the host's memory until it ran out. Past + 10,000 held requests, a new one is now answered at once with `-32000` + "the host is busy: 10000 of this server's requests are already waiting; + send this one again later", and the host logs the first refusal and every + hundredth. A server that doesn't send a refused request again loses it. + Notifications keep their own bound of 1,000, now counted apart from + requests. Before, held requests counted toward it, so a pause holding many + requests dropped notifications it had room for. diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 7f7eb4337..760ef518d 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -295,10 +295,21 @@ export class McplServerConnection extends EventEmitter { private controlPlaneReady = false; private dataPlaneReady = false; private bufferedEvents: Array<{ event: string; args: unknown[] }> = []; - /** Cap on buffered responder-less events during a plane pause (issue #122). - * Held server→host requests are never dropped and may exceed this. */ - private static readonly MAX_BUFFERED_EVENTS = 1000; + /** How many of bufferedEvents are server→host requests, each holding a + * responder its server waits on. Every change to bufferedEvents keeps it. */ + private heldRequests = 0; + /** Cap on held notifications (responder-less events) during a plane pause + * (issue #122), counted apart from requests: past it, the oldest held + * notification is dropped. */ + private static readonly MAX_HELD_NOTIFICATIONS = 1000; + /** Budget of held requests during a plane pause (#387), counted apart from + * notifications: past it, a request is refused at once, never dropped. It + * bounds a server that pipelines requests without waiting for answers, + * while a busy server's ordinary pause, a long quiesce over a busy guild, + * stays under it. */ + private static readonly MAX_HELD_REQUESTS = 10_000; private droppedBufferedEvents = 0; + private refusedHeldRequests = 0; /** True while ready()/readyControlPlane() re-emit a buffer snapshot. Items * re-buffering mid-flush were already admitted once — applying the cap to * them again would evict events that were legally held (the cap targets @@ -375,6 +386,7 @@ export class McplServerConnection extends EventEmitter { // reports that pause's drops, and the next pause counts (and logs) its own. const dropped = this.droppedBufferedEvents; this.droppedBufferedEvents = 0; + this.heldRequests = 0; if (pending.length > 200) { console.error( `[mcpl] ${this.id}: flushing ${pending.length} buffered event(s) synchronously ` + @@ -406,6 +418,7 @@ export class McplServerConnection extends EventEmitter { this.controlPlaneReady = true; const pending = this.bufferedEvents; this.bufferedEvents = []; + this.heldRequests = 0; this.flushDepth++; try { for (const item of pending) { @@ -501,38 +514,57 @@ export class McplServerConnection extends EventEmitter { } return super.emit(event, ...args); } - // Buffer cap (issue #122): a long data-plane pause (quiesce window) on a - // busy server would otherwise grow this array without bound. Only - // responder-LESS events are ever dropped — a buffered server→host REQUEST - // holds a live responder the server is blocking on, and dropping it would - // strand that call forever (the server timing out and reconnecting is the - // accepted bound there, and reconnect re-runs the barrier funnel). - if (this.flushDepth === 0 && this.bufferedEvents.length >= McplServerConnection.MAX_BUFFERED_EVENTS) { - const hasResponder = (item: { args: unknown[] }): boolean => { - const responder = item.args[1] as { respond?: unknown } | undefined; - return typeof responder?.respond === 'function'; - }; - if (!hasResponder({ args })) { + // Held until the plane opens. A long pause (a quiesce, say) on a busy or a + // runaway server would otherwise grow this array without bound, so + // requests and notifications each have a bound, counted apart. A request + // holds a responder its server waits on, and dropping it would strand that + // call, so past MAX_HELD_REQUESTS (#387) it's refused at once with -32000, + // the code of the host's other transient refusals, and its server can send + // it again. A notification has no one to tell, so past + // MAX_HELD_NOTIFICATIONS (issue #122) the oldest held notification is + // dropped to make room. An item held again while a flush re-emits a + // snapshot (flushDepth) was admitted once already, and counts without + // being bounded again. + const responder = args[1] as { respond?: unknown; respondError?: (code: number, message: string) => void } | undefined; + const isRequest = McplServerConnection.hasResponder({ args }); + if (this.flushDepth === 0) { + if (isRequest && this.heldRequests >= McplServerConnection.MAX_HELD_REQUESTS) { + this.refusedHeldRequests++; + if (this.refusedHeldRequests % 100 === 1) { + console.error( + `[mcpl] ${this.id}: held-request budget (${McplServerConnection.MAX_HELD_REQUESTS}) — ` + + `refused ${this.refusedHeldRequests} request(s) with -32000 while the plane is paused`, + ); + } + responder?.respondError?.( + -32000, + `the host is busy: ${McplServerConnection.MAX_HELD_REQUESTS} of this server's requests are already waiting; send this one again later`, + ); + return true; + } + if (!isRequest && this.bufferedEvents.length - this.heldRequests >= McplServerConnection.MAX_HELD_NOTIFICATIONS) { this.droppedBufferedEvents++; if (this.droppedBufferedEvents % 100 === 1) { console.error( - `[mcpl] ${this.id}: buffered-event cap (${McplServerConnection.MAX_BUFFERED_EVENTS}) — ` + + `[mcpl] ${this.id}: buffered-event cap (${McplServerConnection.MAX_HELD_NOTIFICATIONS}) — ` + `dropped ${this.droppedBufferedEvents} responder-less event(s) while the plane is paused`, ); } - const dropIndex = this.bufferedEvents.findIndex((item) => !hasResponder(item)); - if (dropIndex >= 0) { - this.bufferedEvents.splice(dropIndex, 1); // drop-oldest non-request - this.bufferedEvents.push({ event: name, args }); - } - // else: buffer is all held requests — drop the incoming event instead. - return true; + const oldest = this.bufferedEvents.findIndex((item) => !McplServerConnection.hasResponder(item)); + if (oldest >= 0) this.bufferedEvents.splice(oldest, 1); } } this.bufferedEvents.push({ event: name, args }); + if (isRequest) this.heldRequests++; return true; } + /** A buffered server→host request: it carries a responder its server waits on. */ + private static hasResponder(item: { args: unknown[] }): boolean { + const responder = item.args[1] as { respond?: unknown } | undefined; + return typeof responder?.respond === 'function'; + } + /** Event-name form of METHOD_TO_REQUIRED_CAPABILITY, for admission-time * enforcement in emit(). */ private static readonly EVENT_TO_REQUIRED_CAPABILITY: Record = { diff --git a/test/mcpl-held-request-budget.test.ts b/test/mcpl-held-request-budget.test.ts new file mode 100644 index 000000000..b39d03b37 --- /dev/null +++ b/test/mcpl-held-request-budget.test.ts @@ -0,0 +1,109 @@ +/** + * What a connection holds while a plane is paused (#387, issue #122). Requests + * and notifications are bounded apart: past 10,000 held requests a new one is + * refused at once with -32000, never dropped, and past 1,000 held + * notifications the oldest notification is dropped. A server that pipelines + * requests without waiting for answers no longer grows the host's memory + * without bound, and held requests never crowd out notifications. + */ +import { test, type TestContext } from 'node:test'; +import assert from 'node:assert/strict'; + +import { McplServerConnection } from '../src/mcpl/server-connection.js'; +import { McplTransport } from '../src/mcpl/transport.js'; +import { CapabilityGrant, ALL_CAPABILITY_PATHS } from '../src/mcpl/capability-grant.js'; + +const REQUESTS = 10_000; +const NOTIFICATIONS = 1_000; +const BUSY = `the host is busy: ${REQUESTS} of this server's requests are already waiting; send this one again later`; + +class LineTransport extends McplTransport { + constructor() { super(); } + readonly kind = 'stdio' as const; + readonly lines: string[] = []; + writeLine(line: string): void { this.lines.push(line); } + async close(): Promise {} +} + +/** A connection on the real routing, its planes closed as at startup, with a + * push/event listener that records what reaches it. */ +function pausedConnection(t: TestContext) { + const transport = new LineTransport(); + const connection = new (McplServerConnection as any)('srv', null, transport) as McplServerConnection; + connection.establishGrant(new CapabilityGrant(new Set(ALL_CAPABILITY_PATHS), [])); + const delivered: Array = []; + connection.on('push-event', (params: { eventId: string }, responder?: { id: string | number }) => { + delivered.push(responder ? responder.id : params.eventId); + }); + const logged: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }); + let next = 1; + const requests = (count: number) => { + for (let i = 0; i < count; i++) { + const id = next++; + transport.emit('line', JSON.stringify({ jsonrpc: '2.0', id, method: 'push/event', params: { featureSet: 'fs', eventId: `r${id}` } })); + } + }; + const notifications = (count: number, from = 1) => { + for (let i = from; i < from + count; i++) { + transport.emit('line', JSON.stringify({ jsonrpc: '2.0', method: 'push/event', params: { featureSet: 'fs', eventId: `n${i}` } })); + } + }; + const answers = () => transport.lines.map((line) => JSON.parse(line)); + const held = () => (connection as any).bufferedEvents.length as number; + return { connection, delivered, logged, requests, notifications, answers, held }; +} + +test('past 10,000 held requests, a new one is refused at once with -32000; the held ones are delivered in order when the plane opens', (t) => { + const { connection, delivered, logged, requests, answers, held } = pausedConnection(t); + requests(REQUESTS); + assert.equal(held(), REQUESTS); + assert.deepEqual(answers(), []); + requests(2); + assert.deepEqual(answers(), [ + { jsonrpc: '2.0', id: REQUESTS + 1, error: { code: -32000, message: BUSY } }, + { jsonrpc: '2.0', id: REQUESTS + 2, error: { code: -32000, message: BUSY } }, + ]); + assert.equal(held(), REQUESTS); + // One line for the first refusal, then one per hundred. + assert.deepEqual(logged, [`[mcpl] srv: held-request budget (${REQUESTS}) — refused 1 request(s) with -32000 while the plane is paused`]); + connection.ready(); + assert.equal(delivered.length, REQUESTS); + assert.deepEqual(delivered.slice(0, 2), [1, 2]); + assert.equal(delivered.at(-1), REQUESTS); +}); + +test('notifications are counted apart from held requests: with 10,000 requests held, a notification is still held, and 1,000 notifications are kept, newest last', (t) => { + const { connection, delivered, requests, notifications, answers, held } = pausedConnection(t); + requests(REQUESTS); + notifications(NOTIFICATIONS + 1); + // Nothing refused: notifications don't spend the requests' budget. + assert.deepEqual(answers(), []); + // The oldest notification made room for the newest. + assert.equal(held(), REQUESTS + NOTIFICATIONS); + connection.ready(); + const deliveredNotifications = delivered.filter((id) => typeof id === 'string'); + assert.equal(deliveredNotifications.length, NOTIFICATIONS); + assert.equal(deliveredNotifications[0], 'n2'); + assert.equal(deliveredNotifications.at(-1), `n${NOTIFICATIONS + 1}`); +}); + +test('the budget counts what is held: requests held again by a control-plane flush still count, and an open plane frees the budget for the next pause', (t) => { + const { connection, delivered, requests, answers, held } = pausedConnection(t); + requests(REQUESTS / 2); + // The control plane opens and the data plane stays paused, as during a + // reconnect or an awareness drain: every push is held again. + connection.readyControlPlane(); + assert.equal(held(), REQUESTS / 2); + requests(REQUESTS / 2); + assert.deepEqual(answers(), [], 'the half held again and the half sent since fill the budget exactly'); + requests(1); + assert.deepEqual(answers().map((answer) => [answer.id, answer.error?.code]), [[REQUESTS + 1, -32000]]); + connection.ready(); + assert.equal(delivered.length, REQUESTS); + // A quiesce pauses the data plane again; the budget is whole. + connection.pauseDataPlane(); + requests(REQUESTS); + assert.equal(answers().length, 1, 'no new refusal'); + assert.equal(held(), REQUESTS); +}); diff --git a/test/mcpl-request-handler-errors.test.ts b/test/mcpl-request-handler-errors.test.ts index adfbdd081..7e255a375 100644 --- a/test/mcpl-request-handler-errors.test.ts +++ b/test/mcpl-request-handler-errors.test.ts @@ -908,20 +908,20 @@ test('a request form held at the buffer cap is kept, and answered when its plane const errors = t.mock.method(console, 'error', () => {}); const { connection, transport, refreshed } = stagedHarness(); transport.emit('line', JSON.stringify({ jsonrpc: '2.0', id: 5, method: 'mcpl/manifestChanged', params: { revision: 'held' } })); - // 1,100 notifications pass the cap of 1,000 held events by 101. + // 1,100 notifications pass the cap of 1,000 held notifications by 100. The + // request form is counted apart, against the budget for held requests. for (let i = 0; i < 1_100; i++) { transport.emit('line', JSON.stringify({ jsonrpc: '2.0', method: 'mcpl/manifestChanged', params: { revision: `n${i}` } })); } connection.ready(); await settle(); assert.deepEqual(transport.lines.map((line) => JSON.parse(line)), [{ jsonrpc: '2.0', id: 5, result: {} }]); - assert.equal(refreshed.length, 1_000); - assert.deepEqual(refreshed.slice(0, 2), [{ revision: 'held' }, { revision: 'n101' }]); + assert.equal(refreshed.length, 1_001); + assert.deepEqual(refreshed.slice(0, 2), [{ revision: 'held' }, { revision: 'n100' }]); assert.deepEqual(refreshed.at(-1), { revision: 'n1099' }); assert.deepEqual(errors.mock.calls.map((call) => String(call.arguments[0])), [ '[mcpl] srv: buffered-event cap (1000) — dropped 1 responder-less event(s) while the plane is paused', - '[mcpl] srv: buffered-event cap (1000) — dropped 101 responder-less event(s) while the plane is paused', - '[mcpl] srv: flushing 1000 buffered event(s) synchronously (long pause backlog; 101 dropped at cap)', + '[mcpl] srv: flushing 1001 buffered event(s) synchronously (long pause backlog; 100 dropped at cap)', ]); }); From 8758a9dd26070a24f13d1d7c65aacbe594cb4980 Mon Sep 17 00:00:00 2001 From: Tabitha-2696 Date: Fri, 16 Oct 2026 16:51:48 -0400 Subject: [PATCH 2/3] fix(mcpl): each pause logs its first refused request, and the budget's tests pin the logs and a re-hold during ready() (#387) Our internal review of the first commit found these, each fixed here. refusedHeldRequests lived as long as the connection, so after a pause had refused a few requests, a later pause's first refusal wasn't logged (the throttle logs counts 1, 101, 201 ...). ready(), where the planes open and a pause ends, now zeroes it with the held count. MAX_HELD_REQUESTS' doc claimed that a long quiesce over a busy guild stays under the budget, which nothing measured. It now says what the budget is for, and that a server that doesn't resend a refused request loses it. A listener that pauses the data plane during ready()'s own flush holds the rest of the snapshot again, and nothing tested that those count. A new test: two held requests, the first one's handler pauses the plane, the second is held again, and 9,999 more fill the budget. The refusal log's cadence and the drop log had no test. The first test refuses 101 requests and asserts the two lines for refusals 1 and 101; the second asserts the drop line for its one dropped notification; the third asserts that the second pause logs its first refusal. Co-Authored-By: Claude Opus 5.5 --- src/mcpl/server-connection.ts | 10 +++++--- test/mcpl-held-request-budget.test.ts | 37 +++++++++++++++++++++------ 2 files changed, 36 insertions(+), 11 deletions(-) diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 760ef518d..f8e130b74 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -304,11 +304,14 @@ export class McplServerConnection extends EventEmitter { private static readonly MAX_HELD_NOTIFICATIONS = 1000; /** Budget of held requests during a plane pause (#387), counted apart from * notifications: past it, a request is refused at once, never dropped. It - * bounds a server that pipelines requests without waiting for answers, - * while a busy server's ordinary pause, a long quiesce over a busy guild, - * stays under it. */ + * bounds a server that pipelines requests without waiting for answers. A + * server that doesn't send a refused request again loses it (discord-mcpl + * doesn't), so the budget sits far above the notifications' cap, to leave + * a busy server's long pause room. */ private static readonly MAX_HELD_REQUESTS = 10_000; private droppedBufferedEvents = 0; + /** Requests refused at the budget since the planes last opened (ready()), + * so each pause's first refusal is logged. */ private refusedHeldRequests = 0; /** True while ready()/readyControlPlane() re-emit a buffer snapshot. Items * re-buffering mid-flush were already admitted once — applying the cap to @@ -387,6 +390,7 @@ export class McplServerConnection extends EventEmitter { const dropped = this.droppedBufferedEvents; this.droppedBufferedEvents = 0; this.heldRequests = 0; + this.refusedHeldRequests = 0; if (pending.length > 200) { console.error( `[mcpl] ${this.id}: flushing ${pending.length} buffered event(s) synchronously ` + diff --git a/test/mcpl-held-request-budget.test.ts b/test/mcpl-held-request-budget.test.ts index b39d03b37..f74aaef43 100644 --- a/test/mcpl-held-request-budget.test.ts +++ b/test/mcpl-held-request-budget.test.ts @@ -59,14 +59,14 @@ test('past 10,000 held requests, a new one is refused at once with -32000; the h requests(REQUESTS); assert.equal(held(), REQUESTS); assert.deepEqual(answers(), []); - requests(2); - assert.deepEqual(answers(), [ - { jsonrpc: '2.0', id: REQUESTS + 1, error: { code: -32000, message: BUSY } }, - { jsonrpc: '2.0', id: REQUESTS + 2, error: { code: -32000, message: BUSY } }, - ]); + requests(101); + assert.equal(answers().length, 101); + assert.deepEqual(answers()[0], { jsonrpc: '2.0', id: REQUESTS + 1, error: { code: -32000, message: BUSY } }); + assert.deepEqual(answers()[100], { jsonrpc: '2.0', id: REQUESTS + 101, error: { code: -32000, message: BUSY } }); assert.equal(held(), REQUESTS); // One line for the first refusal, then one per hundred. - assert.deepEqual(logged, [`[mcpl] srv: held-request budget (${REQUESTS}) — refused 1 request(s) with -32000 while the plane is paused`]); + assert.deepEqual(logged, [1, 101].map((n) => + `[mcpl] srv: held-request budget (${REQUESTS}) — refused ${n} request(s) with -32000 while the plane is paused`)); connection.ready(); assert.equal(delivered.length, REQUESTS); assert.deepEqual(delivered.slice(0, 2), [1, 2]); @@ -74,11 +74,12 @@ test('past 10,000 held requests, a new one is refused at once with -32000; the h }); test('notifications are counted apart from held requests: with 10,000 requests held, a notification is still held, and 1,000 notifications are kept, newest last', (t) => { - const { connection, delivered, requests, notifications, answers, held } = pausedConnection(t); + const { connection, delivered, logged, requests, notifications, answers, held } = pausedConnection(t); requests(REQUESTS); notifications(NOTIFICATIONS + 1); // Nothing refused: notifications don't spend the requests' budget. assert.deepEqual(answers(), []); + assert.deepEqual(logged, [`[mcpl] srv: buffered-event cap (${NOTIFICATIONS}) — dropped 1 responder-less event(s) while the plane is paused`]); // The oldest notification made room for the newest. assert.equal(held(), REQUESTS + NOTIFICATIONS); connection.ready(); @@ -89,7 +90,7 @@ test('notifications are counted apart from held requests: with 10,000 requests h }); test('the budget counts what is held: requests held again by a control-plane flush still count, and an open plane frees the budget for the next pause', (t) => { - const { connection, delivered, requests, answers, held } = pausedConnection(t); + const { connection, delivered, logged, requests, answers, held } = pausedConnection(t); requests(REQUESTS / 2); // The control plane opens and the data plane stays paused, as during a // reconnect or an awareness drain: every push is held again. @@ -106,4 +107,24 @@ test('the budget counts what is held: requests held again by a control-plane flu requests(REQUESTS); assert.equal(answers().length, 1, 'no new refusal'); assert.equal(held(), REQUESTS); + // This pause's first refusal is logged too. + logged.length = 0; + requests(1); + assert.equal(answers().length, 2); + assert.deepEqual(logged, [`[mcpl] srv: held-request budget (${REQUESTS}) — refused 1 request(s) with -32000 while the plane is paused`]); +}); + +test('a request held again during ready()\'s own flush, when a listener pauses the plane, counts toward the budget', (t) => { + const { connection, delivered, requests, answers, held } = pausedConnection(t); + // The first push's handler pauses the data plane, as a quiesce begun from + // inside a handler would, so the rest of ready()'s snapshot is held again. + connection.once('push-event', () => connection.pauseDataPlane()); + requests(2); + connection.ready(); + assert.deepEqual(delivered, [1]); + assert.equal(held(), 1); + requests(REQUESTS - 1); + assert.deepEqual(answers(), [], 'the one held again and the ones sent since fill the budget exactly'); + requests(1); + assert.deepEqual(answers().map((answer) => [answer.id, answer.error?.code]), [[REQUESTS + 2, -32000]]); }); From ed499917cf1da5d6e2df325f57ed903b2621205b Mon Sep 17 00:00:00 2001 From: Tabitha-2696 Date: Sun, 18 Oct 2026 05:00:11 -0400 Subject: [PATCH 3/3] refactor(mcpl): a flush's held items are bounded as any are, since none can pass the bounds (#387) ready() and readyControlPlane() take the whole buffer and zero its count before they re-emit it, so what a flush holds again was admitted within the held bounds and stays within them. The flushDepth guard, which kept those items from being bounded a second time, could change no outcome: the notification cap has had it since #122, and the request budget took it from there. It goes, with the field and the try/finally that kept its count, and the hold site's comment says why bounding again changes nothing. Co-Authored-By: Claude Opus 5.5 --- src/mcpl/server-connection.ts | 89 ++++++++++++++--------------------- 1 file changed, 36 insertions(+), 53 deletions(-) diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index f8e130b74..1e2612c89 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -313,11 +313,6 @@ export class McplServerConnection extends EventEmitter { /** Requests refused at the budget since the planes last opened (ready()), * so each pause's first refusal is logged. */ private refusedHeldRequests = 0; - /** True while ready()/readyControlPlane() re-emit a buffer snapshot. Items - * re-buffering mid-flush were already admitted once — applying the cap to - * them again would evict events that were legally held (the cap targets - * NEW arrivals during a pause, not churn from control-plane flushes). */ - private flushDepth = 0; // Reconnect state (adapted from Anarchid/agent-framework@mcpl-module-proto) private config: McplServerConfig | null = null; @@ -397,16 +392,11 @@ export class McplServerConnection extends EventEmitter { `(long pause backlog${dropped > 0 ? `; ${dropped} dropped at cap` : ''})`, ); } - this.flushDepth++; - try { - for (const { event, args } of pending) { - // Re-enter the gate for every buffered item. A control handler (notably - // tools/list_changed) may install a newer data-plane barrier while this - // snapshot is being flushed; later data items must observe that pause. - this.dispatchInboundEvent(event, ...args); - } - } finally { - this.flushDepth--; + for (const { event, args } of pending) { + // Re-enter the gate for every buffered item. A control handler (notably + // tools/list_changed) may install a newer data-plane barrier while this + // snapshot is being flushed; later data items must observe that pause. + this.dispatchInboundEvent(event, ...args); } } @@ -423,18 +413,13 @@ export class McplServerConnection extends EventEmitter { const pending = this.bufferedEvents; this.bufferedEvents = []; this.heldRequests = 0; - this.flushDepth++; - try { - for (const item of pending) { - // Re-enter the ADMISSION gate (this.emit), never super.emit: control - // events include channels-register/changed, which are grant-gated — - // the old direct flush bypassed positive-grant enforcement entirely - // (PR #79 review blocker 3). Data-plane items re-buffer in order via - // the same call, since dataPlaneReady is still false. - this.dispatchInboundEvent(item.event, ...item.args); - } - } finally { - this.flushDepth--; + for (const item of pending) { + // Re-enter the ADMISSION gate (this.emit), never super.emit: control + // events include channels-register/changed, which are grant-gated — + // the old direct flush bypassed positive-grant enforcement entirely + // (PR #79 review blocker 3). Data-plane items re-buffer in order via + // the same call, since dataPlaneReady is still false. + this.dispatchInboundEvent(item.event, ...item.args); } } @@ -526,37 +511,35 @@ export class McplServerConnection extends EventEmitter { // the code of the host's other transient refusals, and its server can send // it again. A notification has no one to tell, so past // MAX_HELD_NOTIFICATIONS (issue #122) the oldest held notification is - // dropped to make room. An item held again while a flush re-emits a - // snapshot (flushDepth) was admitted once already, and counts without - // being bounded again. + // dropped to make room. A flush takes the whole buffer and zeroes its + // count, so what it holds again was admitted within these bounds and + // stays within them. const responder = args[1] as { respond?: unknown; respondError?: (code: number, message: string) => void } | undefined; const isRequest = McplServerConnection.hasResponder({ args }); - if (this.flushDepth === 0) { - if (isRequest && this.heldRequests >= McplServerConnection.MAX_HELD_REQUESTS) { - this.refusedHeldRequests++; - if (this.refusedHeldRequests % 100 === 1) { - console.error( - `[mcpl] ${this.id}: held-request budget (${McplServerConnection.MAX_HELD_REQUESTS}) — ` + - `refused ${this.refusedHeldRequests} request(s) with -32000 while the plane is paused`, - ); - } - responder?.respondError?.( - -32000, - `the host is busy: ${McplServerConnection.MAX_HELD_REQUESTS} of this server's requests are already waiting; send this one again later`, + if (isRequest && this.heldRequests >= McplServerConnection.MAX_HELD_REQUESTS) { + this.refusedHeldRequests++; + if (this.refusedHeldRequests % 100 === 1) { + console.error( + `[mcpl] ${this.id}: held-request budget (${McplServerConnection.MAX_HELD_REQUESTS}) — ` + + `refused ${this.refusedHeldRequests} request(s) with -32000 while the plane is paused`, ); - return true; } - if (!isRequest && this.bufferedEvents.length - this.heldRequests >= McplServerConnection.MAX_HELD_NOTIFICATIONS) { - this.droppedBufferedEvents++; - if (this.droppedBufferedEvents % 100 === 1) { - console.error( - `[mcpl] ${this.id}: buffered-event cap (${McplServerConnection.MAX_HELD_NOTIFICATIONS}) — ` + - `dropped ${this.droppedBufferedEvents} responder-less event(s) while the plane is paused`, - ); - } - const oldest = this.bufferedEvents.findIndex((item) => !McplServerConnection.hasResponder(item)); - if (oldest >= 0) this.bufferedEvents.splice(oldest, 1); + responder?.respondError?.( + -32000, + `the host is busy: ${McplServerConnection.MAX_HELD_REQUESTS} of this server's requests are already waiting; send this one again later`, + ); + return true; + } + if (!isRequest && this.bufferedEvents.length - this.heldRequests >= McplServerConnection.MAX_HELD_NOTIFICATIONS) { + this.droppedBufferedEvents++; + if (this.droppedBufferedEvents % 100 === 1) { + console.error( + `[mcpl] ${this.id}: buffered-event cap (${McplServerConnection.MAX_HELD_NOTIFICATIONS}) — ` + + `dropped ${this.droppedBufferedEvents} responder-less event(s) while the plane is paused`, + ); } + const oldest = this.bufferedEvents.findIndex((item) => !McplServerConnection.hasResponder(item)); + if (oldest >= 0) this.bufferedEvents.splice(oldest, 1); } this.bufferedEvents.push({ event: name, args }); if (isRequest) this.heldRequests++;