fix(event-stream): cancel unavailable upstream bodies (#1142)

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

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

@@ -31,6 +31,12 @@ function normalizeHeaders(headers) {
   return { ...headers };
 }
 
+async function cancelResponseBody(response) {
+  if (response?.body && typeof response.body.cancel === 'function') {
+    await response.body.cancel().catch(() => {});
+  }
+}
+
 export function createUpstreamSseReader({
   buildUrl,
   getHeaders = () => ({}),
@@ -116,6 +122,7 @@ export function createUpstreamSseReader({
               status: response?.status ?? 0,
               response,
             });
+            await cancelResponseBody(response);
             await waitForReconnectDelay(reconnectDelayMs, signal);
             continue;
           }

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

@@ -165,6 +165,7 @@ describe('createUpstreamSseReader', () => {
   it('reports unavailable upstream responses and continues reconnecting until stopped', async () => {
     const errors = [];
     let attempt = 0;
+    let unavailableBodyCanceled = false;
     let reader;
 
     reader = createUpstreamSseReader({
@@ -173,7 +174,15 @@ describe('createUpstreamSseReader', () => {
       fetchImpl: async (_url, options) => {
         attempt += 1;
         if (attempt === 1) {
-          return { ok: false, status: 503, body: null };
+          return {
+            ok: false,
+            status: 503,
+            body: {
+              cancel: async () => {
+                unavailableBodyCanceled = true;
+              },
+            },
+          };
         }
 
         return createSseResponse({
@@ -199,6 +208,7 @@ describe('createUpstreamSseReader', () => {
         status: 503,
       }),
     ]);
+    expect(unavailableBodyCanceled).toBe(true);
     expect(attempt).toBe(2);
   });
 });