fix(core): prevent unix pubsub subscriber stalls (#17302)

This fixes the UnixSocketPubSub path where one paused Mastra Code
subscriber could block the broker and every other subscriber after a
large stream part/tool result.

Before, broker publish waited for every remote socket write to drain:

```ts
await writeFrame(client.socket, frame);
await this.flush();
```

After, broker delivery is local-first and remote writes are queued per
client with a bounded buffer:

```ts
this.#deliverLocal(topic, brokerEvent);
this.#enqueueBrokerClientWrite(client, frame);
```

A stuck remote client can now buffer briefly and catch up if it resumes,
but if its per-client queue exceeds the cap we drop only that client.
The active broker and healthy subscribers keep streaming.

I verified this with the focused core tests and a manual 3-instance
Mastra Code repro where one subscriber was paused with `kill -STOP`; the
broker and healthy subscriber kept streaming, then the paused process
resumed cleanly.

---------

Co-authored-by: Mastra Code (openai/gpt-5.5) <noreply@mastra.ai>

Author: TylerBarnes

.changeset/clean-geese-scream.md

@@ -0,0 +1,5 @@
+---
+'@mastra/core': patch
+---
+
+Fixed UnixSocketPubSub streaming so a slow or stuck subscriber no longer blocks active local streams or other subscribers.

packages/core/src/events/unix-socket-pubsub.test.ts

@@ -69,6 +69,117 @@ describe('UnixSocketPubSub', () => {
     expect(secondCb.mock.calls[0]![0].type).toBe('hello');
   });
 
+  it('allows a temporarily backpressured remote client to catch up below the queue cap', async () => {
+    const path = await socketPath();
+    const broker = new UnixSocketPubSub(path, { maxRemoteClientQueuedBytes: 1024 * 1024 });
+    pubsubs.push(broker);
+
+    const brokerCb = vi.fn();
+    await broker.subscribe('topic-a', brokerCb);
+
+    const frames: any[] = [];
+    const rawClient = net.createConnection(path);
+    rawClient.setEncoding('utf8');
+    let buffer = '';
+    rawClient.on('data', (chunk: string) => {
+      buffer += chunk;
+      while (true) {
+        const newlineIndex = buffer.indexOf('\n');
+        if (newlineIndex === -1) break;
+        const line = buffer.slice(0, newlineIndex);
+        buffer = buffer.slice(newlineIndex + 1);
+        if (line.trim()) frames.push(JSON.parse(line));
+      }
+    });
+
+    const waitForRawFrame = async (predicate: (frame: any) => boolean) => {
+      await waitFor(() => {
+        expect(frames.some(predicate)).toBe(true);
+      });
+    };
+
+    try {
+      await new Promise<void>((resolve, reject) => {
+        rawClient.once('connect', resolve);
+        rawClient.once('error', reject);
+      });
+      await new Promise<void>((resolve, reject) => {
+        rawClient.write(`${JSON.stringify({ type: 'subscribe', topic: 'topic-a' })}\n`, (error?: Error | null) => {
+          if (error) reject(error);
+          else resolve();
+        });
+      });
+      await waitForRawFrame(frame => frame.type === 'subscribed' && frame.topic === 'topic-a');
+      rawClient.pause();
+
+      const payload = 'x'.repeat(16 * 1024);
+      for (let i = 0; i < 4; i++) {
+        await broker.publish('topic-a', makeEvent({ type: `recover-${i}`, data: { payload } }));
+      }
+
+      expect(broker.remoteClientCount).toBe(1);
+      expect(brokerCb).toHaveBeenCalledTimes(4);
+      rawClient.resume();
+
+      await waitForRawFrame(frame => frame.type === 'event' && frame.event?.type === 'recover-3');
+      expect(broker.remoteClientCount).toBe(1);
+    } finally {
+      rawClient.destroy();
+    }
+  });
+
+  it('does not let a backpressured remote client block local or healthy subscribers', async () => {
+    const path = await socketPath();
+    const broker = new UnixSocketPubSub(path, { maxRemoteClientQueuedBytes: 256 * 1024 });
+    const healthy = new UnixSocketPubSub(path, { maxRemoteClientQueuedBytes: 256 * 1024 });
+    pubsubs.push(broker, healthy);
+
+    const brokerCb = vi.fn();
+    const healthyCb = vi.fn();
+    await broker.subscribe('topic-a', brokerCb);
+    await healthy.subscribe('topic-a', healthyCb);
+
+    const stuck = net.createConnection(path);
+    try {
+      await new Promise<void>((resolve, reject) => {
+        stuck.once('connect', resolve);
+        stuck.once('error', reject);
+      });
+      await new Promise<void>((resolve, reject) => {
+        stuck.write(`${JSON.stringify({ type: 'subscribe', topic: 'topic-a' })}\n`, (error?: Error | null) => {
+          if (error) reject(error);
+          else resolve();
+        });
+      });
+      stuck.pause();
+
+      await waitFor(() => {
+        expect(broker.remoteClientCount).toBe(2);
+      });
+      await new Promise(resolve => setTimeout(resolve, 0));
+
+      const payload = 'x'.repeat(32 * 1024);
+      for (let i = 0; i < 20; i++) {
+        const result = await Promise.race([
+          broker.publish('topic-a', makeEvent({ type: `large-${i}`, data: { payload } })).then(() => 'published'),
+          new Promise(resolve => setTimeout(() => resolve('timeout'), 500)),
+        ]);
+        expect(result).toBe('published');
+        await new Promise(resolve => setTimeout(resolve, 0));
+      }
+
+      expect(brokerCb).toHaveBeenCalledTimes(20);
+      await waitFor(() => {
+        expect(healthyCb.mock.calls.some(call => call[0].type === 'large-19')).toBe(true);
+      });
+      await waitFor(() => {
+        expect(broker.remoteClientCount).toBe(1);
+      });
+    } finally {
+      stuck.destroy();
+    }
+  });
+
   it('isolates local subscriber failures from other subscribers', async () => {
     const path = await socketPath();
     const pubsub = new UnixSocketPubSub(path);

packages/core/src/events/unix-socket-pubsub.ts

@@ -17,40 +17,88 @@ type ClientFrame =
 
 type ServerFrame = { type: 'event'; topic: string; event: Event } | { type: 'subscribed'; topic: string };
 
+type UnixSocketPubSubOptions = {
+  maxRemoteClientQueuedBytes?: number;
+};
+
 type BrokerClient = {
   socket: net.Socket;
   subscriptions: Set<string>;
+  writeChain: Promise<void>;
+  queuedBytes: number;
 };
 
 type SubscribeWaiter = {
   resolve: () => void;
   reject: (error: Error) => void;
 };
 
-function writeFrame(socket: net.Socket, frame: ClientFrame | ServerFrame): Promise<void> {
+const DEFAULT_MAX_REMOTE_CLIENT_QUEUED_BYTES = 64 * 1024 * 1024;
+
+function serializeFrame(frame: ClientFrame | ServerFrame): string {
+  return `${JSON.stringify(frame)}\n`;
+}
+
+function writeSerializedFrame(socket: net.Socket, serializedFrame: string): Promise<void> {
   return new Promise((resolve, reject) => {
-    const onError = (error: Error) => {
+    let writeCompleted = false;
+    let drainCompleted = true;
+    let settled = false;
+
+    const cleanup = () => {
+      socket.off('error', onError);
+      socket.off('close', onClose);
       socket.off('drain', onDrain);
-      reject(error);
     };
-    const onDrain = () => {
-      socket.off('error', onError);
+    const settle = (error?: Error) => {
+      if (settled) return;
+      settled = true;
+      cleanup();
+      if (error) {
+        reject(error);
+        return;
+      }
       resolve();
     };
+    const maybeResolve = () => {
+      if (writeCompleted && drainCompleted) {
+        settle();
+      }
+    };
+    const onError = (error: Error) => settle(error);
+    const onClose = () => settle(new Error('UnixSocketPubSub socket closed before write completed'));
+    const onDrain = () => {
+      drainCompleted = true;
+      maybeResolve();
+    };
 
     socket.once('error', onError);
-    const drained = socket.write(`${JSON.stringify(frame)}\n`, () => {
-      if (drained) {
-        socket.off('error', onError);
-        resolve();
-      }
-    });
+    socket.once('close', onClose);
+    let drained: boolean;
+    try {
+      drained = socket.write(serializedFrame, error => {
+        if (error) {
+          settle(error);
+          return;
+        }
+        writeCompleted = true;
+        maybeResolve();
+      });
+    } catch (error) {
+      settle(error as Error);
+      return;
+    }
     if (!drained) {
+      drainCompleted = false;
       socket.once('drain', onDrain);
     }
   });
 }
 
+function writeFrame(socket: net.Socket, frame: ClientFrame | ServerFrame): Promise<void> {
+  return writeSerializedFrame(socket, serializeFrame(frame));
+}
+
 function nextTick(): Promise<void> {
   return new Promise(resolve => setImmediate(resolve));
 }
@@ -85,12 +133,14 @@ export class UnixSocketPubSub extends PubSub {
   #callbacks = new Map<string, Set<EventCallback>>();
   #subscribeWaiters = new Map<string, SubscribeWaiter[]>();
   #brokerClients = new Map<net.Socket, BrokerClient>();
-  #pendingWrites: Promise<void>[] = [];
+  #pendingWrites = new Set<Promise<void>>();
   #recovering?: Promise<void>;
+  #maxRemoteClientQueuedBytes: number;
 
-  constructor(socketPath: string) {
+  constructor(socketPath: string, options: UnixSocketPubSubOptions = {}) {
     super();
     this.socketPath = socketPath;
+    this.#maxRemoteClientQueuedBytes = options.maxRemoteClientQueuedBytes ?? DEFAULT_MAX_REMOTE_CLIENT_QUEUED_BYTES;
   }
 
   override get supportedModes(): ReadonlyArray<PubSubDeliveryMode> {
@@ -160,8 +210,7 @@ export class UnixSocketPubSub extends PubSub {
   }
 
   async flush(): Promise<void> {
-    await Promise.allSettled(this.#pendingWrites);
-    this.#pendingWrites = [];
+    await Promise.allSettled([...this.#pendingWrites]);
   }
 
   async close(): Promise<void> {
@@ -172,10 +221,9 @@ export class UnixSocketPubSub extends PubSub {
     this.#clientSocket = undefined;
     this.#rejectSubscribeWaiters(new Error('UnixSocketPubSub is closed'));
 
-    for (const client of this.#brokerClients.values()) {
-      client.socket.destroy();
+    for (const client of [...this.#brokerClients.values()]) {
+      this.#removeBrokerClient(client);
     }
-    this.#brokerClients.clear();
 
     if (this.#server) {
       await new Promise<void>(resolve => this.#server?.close(() => resolve()));
@@ -436,21 +484,67 @@ export class UnixSocketPubSub extends PubSub {
   }
 
   #handleBrokerClient(socket: net.Socket) {
-    const client: BrokerClient = { socket, subscriptions: new Set() };
+    const client: BrokerClient = {
+      socket,
+      subscriptions: new Set(),
+      writeChain: Promise.resolve(),
+      queuedBytes: 0,
+    };
     this.#brokerClients.set(socket, client);
     readFrames(socket, frame => {
       const clientFrame = frame as ClientFrame;
       if (clientFrame.type === 'subscribe') {
         client.subscriptions.add(clientFrame.topic);
-        void writeFrame(socket, { type: 'subscribed', topic: clientFrame.topic }).catch(() => {});
+        this.#enqueueBrokerClientWrite(client, { type: 'subscribed', topic: clientFrame.topic });
       } else if (clientFrame.type === 'unsubscribe') {
         client.subscriptions.delete(clientFrame.topic);
       } else if (clientFrame.type === 'publish') {
         void this.#publishFromBroker(clientFrame.topic, clientFrame.event);
       }
     });
-    socket.on('close', () => this.#brokerClients.delete(socket));
-    socket.on('error', () => this.#brokerClients.delete(socket));
+    socket.on('close', () => this.#removeBrokerClient(client));
+    socket.on('error', () => this.#removeBrokerClient(client));
+  }
+
+  #enqueueBrokerClientWrite(client: BrokerClient, frame: ServerFrame) {
+    if (this.#brokerClients.get(client.socket) !== client || client.socket.destroyed) return;
+
+    const serializedFrame = serializeFrame(frame);
+    const queuedBytes = Buffer.byteLength(serializedFrame);
+    if (client.queuedBytes + queuedBytes > this.#maxRemoteClientQueuedBytes) {
+      this.#removeBrokerClient(client);
+      return;
+    }
+
+    client.queuedBytes += queuedBytes;
+
+    const write = client.writeChain
+      .catch(() => {})
+      .then(async () => {
+        if (this.#brokerClients.get(client.socket) !== client || client.socket.destroyed) return;
+        await writeSerializedFrame(client.socket, serializedFrame);
+      })
+      .catch(() => {
+        this.#removeBrokerClient(client);
+      })
+      .finally(() => {
+        client.queuedBytes = Math.max(0, client.queuedBytes - queuedBytes);
+      });
+
+    client.writeChain = write;
+    this.#pendingWrites.add(write);
+    void write.finally(() => this.#pendingWrites.delete(write));
+  }
+
+  #removeBrokerClient(client: BrokerClient) {
+    if (this.#brokerClients.get(client.socket) !== client) return;
+    this.#brokerClients.delete(client.socket);
+    client.subscriptions.clear();
+    client.queuedBytes = 0;
+    client.writeChain = Promise.resolve();
+    if (!client.socket.destroyed) {
+      client.socket.destroy();
+    }
   }
 
   #handleServerFrame(frame: ServerFrame) {
@@ -484,10 +578,8 @@ export class UnixSocketPubSub extends PubSub {
       if (!client.subscriptions.has(topic) || client.socket.destroyed) continue;
       // Lazily build the frame only when we know at least one client needs it.
       frame ??= { type: 'event', topic, event: brokerEvent };
-      const write = writeFrame(client.socket, frame).catch(() => {});
-      this.#pendingWrites.push(write);
+      this.#enqueueBrokerClientWrite(client, frame);
     }
-    if (frame) await this.flush();
   }
 
   #deliverLocal(topic: string, event: Event) {
@@ -510,14 +602,39 @@ export class UnixSocketPubSub extends PubSub {
   }
 
   async #sendToBroker(frame: ClientFrame) {
+    try {
+      await this.#sendToActiveBroker(frame);
+    } catch (error) {
+      if (this.#closed) throw error;
+      const failedSocket = this.#clientSocket;
+      this.#clientSocket = undefined;
+      failedSocket?.destroy();
+      await this.#ensureStarted(true);
+      await this.#sendToActiveBroker(frame);
+    }
+  }
+
+  async #sendToActiveBroker(frame: ClientFrame) {
     const socket = this.#clientSocket;
     if (!socket || socket.destroyed) {
       await this.#ensureStarted(true);
     }
+    if (this.#isBroker) {
+      await this.#handlePromotedBrokerFrame(frame);
+      return;
+    }
     const activeSocket = this.#clientSocket;
     if (!activeSocket || activeSocket.destroyed) {
       throw new Error('UnixSocketPubSub is not connected to a broker');
     }
     await writeFrame(activeSocket, frame);
   }
+
+  async #handlePromotedBrokerFrame(frame: ClientFrame) {
+    if (frame.type === 'subscribe') {
+      this.#settleSubscribeWaiters(frame.topic);
+    } else if (frame.type === 'publish') {
+      await this.#publishFromBroker(frame.topic, frame.event);
+    }
+  }
 }