fix: clean up upstream reader abort listener (#1378)

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

packages/web/server/lib/event-stream/upstream-reader.js

@@ -63,21 +63,34 @@ export function createUpstreamSseReader({
   let stopped = false;
   let activeController = null;
   let lastEventId = typeof initialLastEventId === 'string' ? initialLastEventId : '';
+  let stopListenerAttached = false;
 
-  const stop = () => {
+  function detachStopListener() {
+    if (!stopListenerAttached) return;
+    signal?.removeEventListener('abort', stop);
+    stopListenerAttached = false;
+  }
+
+  function attachStopListener() {
+    if (!signal || signal.aborted || stopListenerAttached) return;
+    signal.addEventListener('abort', stop, { once: true });
+    stopListenerAttached = true;
+  }
+
+  function stop() {
     stopped = true;
+    detachStopListener();
     if (activeController && !activeController.signal.aborted) {
       activeController.abort();
     }
-  };
-
-  signal?.addEventListener('abort', stop, { once: true });
+  }
 
   const start = () => {
     if (running) {
       return running;
     }
 
+    attachStopListener();
     stopped = false;
     running = (async () => {
       while (!stopped && !signal?.aborted) {
@@ -210,6 +223,7 @@ export function createUpstreamSseReader({
         }
       }
     })().finally(() => {
+      detachStopListener();
       running = null;
     });
 

packages/web/server/lib/event-stream/upstream-reader.test.js

@@ -234,7 +234,7 @@ describe('createUpstreamSseReader', () => {
     expect(attempt).toBe(2);
   });
 
-  it('removes reconnect delay abort listeners after normal timeout completion', async () => {
+  it('removes abort listeners after stop', async () => {
     const tracked = createTrackedSignal();
     let attempt = 0;
     let reader;
@@ -270,7 +270,6 @@ describe('createUpstreamSseReader', () => {
     await reader.start();
 
     expect(attempt).toBe(2);
-    // The top-level stop listener remains; the reconnect-delay listener should be removed.
-    expect(tracked.getListenerCount()).toBe(1);
+    expect(tracked.getListenerCount()).toBe(0);
   });
 });