fix(mcpl): held requests get a budget of 10,000, apart from notifications; past it a request is refused (#387)

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.

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<string, string> = {

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<void> {}
+}
+
+/** 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<string | number | undefined> = [];
+  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]]);
+});

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)',
   ]);
 });