Gracefully handle event stream reading errors and cancellations (#1121)

Author: enisdenjo

.changeset/twenty-beans-grab.md

@@ -0,0 +1,5 @@
+---
+'@graphql-tools/executor-http': patch
+---
+
+Gracefully handle event stream reading errors and cancellations

packages/executors/http/src/handleEventStreamResponse.ts

@@ -1,4 +1,8 @@
-import { ExecutionResult, inspect } from '@graphql-tools/utils';
+import {
+  createGraphQLError,
+  ExecutionResult,
+  inspect,
+} from '@graphql-tools/utils';
 import { Repeater } from '@repeaterjs/repeater';
 import { TextDecoder } from '@whatwg-node/fetch';
 import { createResultForAbort } from './utils';
@@ -48,7 +52,24 @@ export function handleEventStreamResponse(
       if (!body?.locked) {
         return stop();
       }
-      const { done, value: chunk } = await reader.read();
+      let done: boolean, chunk: Uint8Array<ArrayBufferLike> | undefined;
+      try {
+        const result = await reader.read();
+        done = result.done;
+        chunk = result.value;
+      } catch (err) {
+        if (signal?.aborted) {
+          await push(createResultForAbort(signal.reason));
+          return stop();
+        }
+        const errErr = err instanceof Error ? err : new Error(String(err));
+        await push({
+          errors: [
+            createGraphQLError(errErr.message, { originalError: errErr }),
+          ],
+        });
+        return stop();
+      }
       if (done) {
         return stop();
       }

packages/executors/http/tests/handleEventStreamResponse.test.ts

@@ -119,4 +119,102 @@ describe('handleEventStreamResponse', () => {
       value: undefined,
     });
   });
+
+  it.skipIf(
+    // we skip bun because we cant cancel the stream while reading it (it's locked)
+    // however, the same test from nodejs applies in bun
+    globalThis.Bun,
+  )('should gracefully report stream cancel with aborted signal', async () => {
+    const ctrl = new AbortController();
+    const readableStream = new ReadableStream<Uint8Array>({
+      start() {
+        // dont enqueue anything, to hang on iterator.next()
+      },
+    });
+
+    const response = new Response(readableStream);
+    const asyncIterable = handleEventStreamResponse(
+      response,
+      undefined,
+      ctrl.signal,
+    );
+    const iterator = asyncIterable[Symbol.asyncIterator]();
+
+    Promise.resolve().then(() => {
+      ctrl.abort(); // we abort
+      readableStream.cancel(); // then cancel
+      // so that the error reported is the abort error
+    });
+
+    await expect(iterator.next()).resolves.toMatchInlineSnapshot(`
+      {
+        "done": false,
+        "value": {
+          "errors": [
+            [GraphQLError: This operation was aborted],
+          ],
+        },
+      }
+    `);
+
+    await expect(iterator.next()).resolves.toMatchInlineSnapshot(`
+      {
+        "done": true,
+        "value": undefined,
+      }
+    `);
+
+    await expect(iterator.return()).resolves.toMatchInlineSnapshot(`
+      {
+        "done": true,
+        "value": undefined,
+      }
+    `);
+  });
+
+  it.skipIf(
+    // we skip bun because we cant cancel the stream while reading it (it's locked)
+    // however, the same test from nodejs applies in bun
+    globalThis.Bun,
+  )('should gracefully report stream errors', async () => {
+    const readableStream = new ReadableStream<Uint8Array>({
+      start() {
+        // dont enqueue anything, to hang on iterator.next()
+      },
+    });
+
+    const response = new Response(readableStream);
+    const asyncIterable = handleEventStreamResponse(response);
+    const iterator = asyncIterable[Symbol.asyncIterator]();
+
+    const originalError = new Error('Oops!');
+    Promise.resolve().then(() => {
+      readableStream.cancel(originalError); // this will throw in reader.read()
+    });
+
+    const { value, done } = await iterator.next();
+    expect(done).toBeFalsy();
+    expect(value).toMatchInlineSnapshot(`
+      {
+        "errors": [
+          [GraphQLError: Oops!],
+        ],
+      }
+    `);
+    expect(value.errors[0].originalError).toBe(originalError);
+
+    await expect(iterator.next()).resolves.toMatchInlineSnapshot(`
+      {
+        "done": true,
+        "value": undefined,
+      }
+    `);
+
+    await expect(iterator.return()).resolves.toMatchInlineSnapshot(`
+      {
+        "done": true,
+        "value": undefined,
+      }
+    `);
+  });
 });