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..1e2612c89 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -295,15 +295,24 @@ 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. 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; - /** 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; + /** Requests refused at the budget since the planes last opened (ready()), + * so each pause's first refusal is logged. */ + private refusedHeldRequests = 0; // Reconnect state (adapted from Anarchid/agent-framework@mcpl-module-proto) private config: McplServerConfig | null = null; @@ -375,22 +384,19 @@ 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; + this.refusedHeldRequests = 0; if (pending.length > 200) { console.error( `[mcpl] ${this.id}: flushing ${pending.length} buffered event(s) synchronously ` + `(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); } } @@ -406,18 +412,14 @@ export class McplServerConnection extends EventEmitter { this.controlPlaneReady = true; const pending = this.bufferedEvents; this.bufferedEvents = []; - 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--; + this.heldRequests = 0; + 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); } } @@ -501,38 +503,55 @@ 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 })) { - this.droppedBufferedEvents++; - if (this.droppedBufferedEvents % 100 === 1) { - console.error( - `[mcpl] ${this.id}: buffered-event cap (${McplServerConnection.MAX_BUFFERED_EVENTS}) — ` + - `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; + // 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. 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 (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_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++; 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..f74aaef43 --- /dev/null +++ b/test/mcpl-held-request-budget.test.ts @@ -0,0 +1,130 @@ +/** + * 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(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, [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]); + 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, 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(); + 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, 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. + 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); + // 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]]); +}); 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)', ]); });