fix(client): settle() drops _listenState by the captured id, not via parked

When a server-cancel (or any termination) is delivered synchronously
inside _parkRequest's send — in-process transports — settle() runs before
`parked` is assigned, so the previous `_listenState.delete(parked.messageId)`
was skipped and the entry registered by onBeforeSend leaked. The catch-up
after _parkRequest only called parked.unpark(), not the delete.

Capture the messageId in the onBeforeSend closure as `listenMessageId`;
settle() and wireTeardown() key off that, so the entry is dropped on every
exit path regardless of whether `parked` has been assigned yet.

Author: felixweinberger

packages/client/src/client/client.ts

@@ -1274,6 +1274,12 @@ export class Client extends Protocol<ClientContext> {
         // before-ack / close-before-ack hangs are impossible by construction.
         let state: 'opening' | 'open' | 'closed' = 'opening';
         let parked: ReturnType<typeof this._parkRequest> | undefined;
+        // The listen request id, captured by `onBeforeSend` BEFORE the request
+        // goes out. `settle()` deletes `_listenState` by this id (not via
+        // `parked.messageId`), so a synchronously-delivered termination during
+        // `_parkRequest`'s send — when `parked` is still unassigned — does not
+        // leak the entry.
+        let listenMessageId: number | undefined;
         let ackTimer: ReturnType<typeof setTimeout> | undefined;
         let resolveOpening!: (honored: SubscriptionFilter) => void;
         let rejectOpening!: (error: Error) => void;
@@ -1297,10 +1303,10 @@ export class Client extends Protocol<ClientContext> {
                 return;
             }
             state = 'closed';
-            if (parked !== undefined) {
-                this._listenState.delete(parked.messageId);
-                parked.unpark();
+            if (listenMessageId !== undefined) {
+                this._listenState.delete(listenMessageId);
             }
+            parked?.unpark();
             if (wasOpening) {
                 rejectOpening(
                     outcome === 'closed'
@@ -1321,7 +1327,7 @@ export class Client extends Protocol<ClientContext> {
         // cancelled).
         const wireTeardown = async (): Promise<void> => {
             requestAbort.abort();
-            const id = parked?.messageId;
+            const id = listenMessageId;
             if (id !== undefined) {
                 await this.transport
                     ?.send({ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: id } })
@@ -1359,6 +1365,7 @@ export class Client extends Protocol<ClientContext> {
                 },
                 { requestSignal: requestAbort.signal },
                 messageId => {
+                    listenMessageId = messageId;
                     this._listenState.set(messageId, {
                         onAck: honored => settle({ ack: honored }),
                         onServerCancel: () => {
@@ -1369,8 +1376,10 @@ export class Client extends Protocol<ClientContext> {
             );
             // A synchronously-delivered termination during `send()` (an
             // in-process transport) ran `settle()` before `parked` was
-            // assigned — unpark now so the handler does not leak. (Cast: TS
-            // control-flow narrowing does not track closure mutation.)
+            // assigned; `settle()` already cleared `_listenState` via
+            // `listenMessageId` — unpark now so the response handler does not
+            // leak either. (Cast: TS control-flow narrowing does not track
+            // closure mutation.)
             if ((state as 'opening' | 'open' | 'closed') === 'closed') parked.unpark();
             // Pre-ack capacity / params rejection arrives as a JSON-RPC error
             // for the listen id; transport close is delivered the same way.

packages/client/test/client/listen.test.ts

@@ -228,6 +228,42 @@ describe('Client.listen()', () => {
         await client.close();
     });
 
+    it('a synchronously-delivered server-cancel during send does not leak a _listenState entry', async () => {
+        // In-process delivery: the server's notifications/cancelled arrives
+        // inside `_parkRequest`'s send (before `parked` is assigned). settle()
+        // must still drop the `_listenState` entry it registered via
+        // onBeforeSend.
+        const [clientTx, serverTx] = InMemoryTransport.createLinkedPair();
+        serverTx.onmessage = m => {
+            const req = m as { id?: number | string; method?: string };
+            if (req.method === 'server/discover' && req.id !== undefined) {
+                void serverTx.send({
+                    jsonrpc: '2.0',
+                    id: req.id,
+                    result: {
+                        resultType: 'complete',
+                        supportedVersions: [MODERN],
+                        capabilities: {},
+                        serverInfo: { name: 's', version: '1' }
+                    }
+                });
+            }
+            if (req.method === 'subscriptions/listen' && req.id !== undefined) {
+                void serverTx.send({ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: req.id } });
+            }
+        };
+        await serverTx.start();
+        const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } });
+        await client.connect(clientTx);
+        const listenState = (client as unknown as { _listenState: Map<unknown, unknown> })._listenState;
+        const before = listenState.size;
+        const error = await client.listen({ toolsListChanged: true }).catch(e => e as Error);
+        expect((error as Error).message).toContain('server cancelled before acknowledging');
+        // No leaked _listenState entry for the listen id.
+        expect(listenState.size).toBe(before);
+        await client.close();
+    });
+
     it('a synchronous transport.send throw does not leak a _responseHandlers entry', async () => {
         const { clientTx } = await scriptedModern();
         const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } });
@@ -242,6 +278,9 @@ describe('Client.listen()', () => {
         expect((error as Error).message).toContain('send blew up');
         // The park primitive unregistered before rethrowing — no leak.
         expect(handlers.size).toBe(before);
+        // settle() in the catch path also dropped the _listenState entry that
+        // onBeforeSend registered before send threw.
+        expect((client as unknown as { _listenState: Map<unknown, unknown> })._listenState.size).toBe(0);
         clientTx.send = realSend;
         await client.close();
     });