fix(event-stream): isolate subscriber failures (#1235)

* fix(event-stream): isolate subscriber failures

* fix(event-stream): handle async subscriber failures

---------

Co-authored-by: Isaac Sanchez <isanchez-hawkins@arize.com>

packages/web/server/lib/event-stream/global-hub.js

@@ -22,9 +22,22 @@ export function createGlobalMessageStreamHub({
   let everConnected = false;
   let buildUrlFailed = false;
 
+  const notifySubscriber = (kind, subscriber, payload) => {
+    try {
+      const result = subscriber(payload);
+      if (result && typeof result.catch === 'function') {
+        result.catch((error) => {
+          console.warn(`Global message stream ${kind} subscriber failed:`, error);
+        });
+      }
+    } catch (error) {
+      console.warn(`Global message stream ${kind} subscriber failed:`, error);
+    }
+  };
+
   const notifyStatus = (status) => {
     for (const subscriber of Array.from(statusSubscribers)) {
-      subscriber(status);
+      notifySubscriber('status', subscriber, status);
     }
   };
 
@@ -81,7 +94,7 @@ export function createGlobalMessageStreamHub({
         }
 
         for (const subscriber of Array.from(eventSubscribers)) {
-          subscriber(normalized);
+          notifySubscriber('event', subscriber, normalized);
         }
       },
       onError(error) {

packages/web/server/lib/event-stream/global-hub.test.js

@@ -0,0 +1,140 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { createGlobalMessageStreamHub } from './global-hub.js';
+
+function createSseResponse({ blocks = [] } = {}) {
+  const encoder = new TextEncoder();
+  let index = 0;
+
+  return {
+    ok: true,
+    body: {
+      getReader() {
+        return {
+          async read() {
+            if (index < blocks.length) {
+              return { value: encoder.encode(blocks[index++]), done: false };
+            }
+            return { value: undefined, done: true };
+          },
+        };
+      },
+    },
+  };
+}
+
+async function waitForAssertion(assertion) {
+  const deadline = Date.now() + 1000;
+  let lastError;
+
+  while (Date.now() < deadline) {
+    try {
+      assertion();
+      return;
+    } catch (error) {
+      lastError = error;
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+  }
+
+  throw lastError;
+}
+
+describe('createGlobalMessageStreamHub', () => {
+  it('continues fanout when an event subscriber throws', async () => {
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
+    const received = [];
+    const hub = createGlobalMessageStreamHub({
+      buildOpenCodeUrl: (pathname) => `http://127.0.0.1:4096${pathname}`,
+      getOpenCodeAuthHeaders: () => ({}),
+      upstreamReconnectDelayMs: 100,
+      fetchImpl: async () => createSseResponse({
+        blocks: [
+          'id: evt-1\ndata: {"type":"session.updated","properties":{}}\n\n',
+        ],
+      }),
+    });
+
+    hub.subscribeEvent(() => {
+      throw new Error('subscriber failed');
+    });
+    hub.subscribeEvent((event) => {
+      received.push(event.eventId);
+    });
+
+    try {
+      hub.start();
+      await waitForAssertion(() => {
+        expect(received).toEqual(['evt-1']);
+      });
+      expect(warnSpy).toHaveBeenCalled();
+    } finally {
+      hub.stop();
+      warnSpy.mockRestore();
+    }
+  });
+
+  it('continues status fanout when a status subscriber throws', async () => {
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
+    const received = [];
+    const hub = createGlobalMessageStreamHub({
+      buildOpenCodeUrl: (pathname) => `http://127.0.0.1:4096${pathname}`,
+      getOpenCodeAuthHeaders: () => ({}),
+      upstreamReconnectDelayMs: 100,
+      fetchImpl: async () => createSseResponse(),
+    });
+
+    hub.subscribeStatus(() => {
+      throw new Error('status subscriber failed');
+    });
+    hub.subscribeStatus((status) => {
+      received.push(status.type);
+    });
+
+    try {
+      hub.start();
+      await waitForAssertion(() => {
+        expect(received).toContain('connect');
+      });
+      expect(warnSpy).toHaveBeenCalled();
+    } finally {
+      hub.stop();
+      warnSpy.mockRestore();
+    }
+  });
+
+  it('continues fanout when an async event subscriber rejects', async () => {
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
+    const received = [];
+    const hub = createGlobalMessageStreamHub({
+      buildOpenCodeUrl: (pathname) => `http://127.0.0.1:4096${pathname}`,
+      getOpenCodeAuthHeaders: () => ({}),
+      upstreamReconnectDelayMs: 100,
+      fetchImpl: async () => createSseResponse({
+        blocks: [
+          'id: evt-1\ndata: {"type":"session.updated","properties":{}}\n\n',
+        ],
+      }),
+    });
+
+    hub.subscribeEvent(async () => {
+      throw new Error('async subscriber failed');
+    });
+    hub.subscribeEvent((event) => {
+      received.push(event.eventId);
+    });
+
+    try {
+      hub.start();
+      await waitForAssertion(() => {
+        expect(received).toEqual(['evt-1']);
+      });
+      await waitForAssertion(() => {
+        expect(warnSpy).toHaveBeenCalled();
+      });
+    } finally {
+      hub.stop();
+      warnSpy.mockRestore();
+    }
+  });
+});