HttpLink/BatchHttpLink: Abort AbortController signal more granularly.

.changeset/smooth-goats-cheat.md

@@ -0,0 +1,13 @@
+---
+'@apollo/client': minor
+---
+
+`HttpLink`/`BatchHttpLink`: Abort AbortController signal more granularly.
+Before this change, when `HttpLink`/`BatchHttpLink` created an `AbortController`
+internally, that would always be `.abort`ed after the request was finished in any
+way.
+This could cause issues with Sentry Session Replay and Next.js App Router Cache
+invalidations, which just replayed the fetch with the same options - including the
+cancelled `AbortSignal`.
+With this change, the `AbortController` will only be `.abort()`ed by outside events,
+not as a consequence of the request finishing or erroring.

src/link/batch-http/batchHttpLink.ts

@@ -16,7 +16,6 @@ import {
   selectHttpOptionsAndBodyInternal,
   defaultPrinter,
   fallbackHttpConfig,
-  createSignalIfSupported,
 } from '../http';
 import { BatchLink } from '../batch';
 import { filterOperationVariables } from "../utils/filterOperationVariables";
@@ -157,11 +156,10 @@ export class BatchHttpLink extends ApolloLink {
         return fromError<FetchResult[]>(parseError);
       }
 
-      let controller: any;
-      if (!(options as any).signal) {
-        const { controller: _controller, signal } = createSignalIfSupported();
-        controller = _controller;
-        if (controller) (options as any).signal = signal;
+      let controller: AbortController | undefined;
+      if (!options.signal && typeof AbortController !== 'undefined') {
+        controller = new AbortController();
+        options.signal = controller.signal;
       }
 
       return new Observable<FetchResult[]>(observer => {
@@ -173,12 +171,14 @@ export class BatchHttpLink extends ApolloLink {
           })
           .then(parseAndCheckHttpResponse(operations))
           .then(result => {
+            controller = undefined;
             // we have data and can send it to back up the link chain
             observer.next(result);
             observer.complete();
             return result;
           })
           .catch(err => {
+            controller = undefined;
             // fetch was cancelled so its already been cleaned up in the unsubscribe
             if (err.name === 'AbortError') return;
             // if it is a network error, BUT there is graphql result info

src/link/http/__tests__/HttpLink.ts

@@ -13,7 +13,7 @@ import { HttpLink } from '../HttpLink';
 import { createHttpLink } from '../createHttpLink';
 import { ClientParseError } from '../serializeFetchParameter';
 import { ServerParseError } from '../parseAndCheckHttpResponse';
-import { ServerError } from '../../..';
+import { FetchResult, ServerError } from '../../..';
 import { voidFetchDuringEachTest } from './helpers';
 import { itAsync } from '../../../testing';
 
@@ -1325,46 +1325,96 @@ describe('HttpLink', () => {
         }),
       );
     });
-    itAsync('supports being cancelled and does not throw', (resolve, reject) => {
-      let called = false;
-      class AbortController {
+
+    describe('AbortController', () => {
+      let aborted = false;
+      let created = false;
+      beforeEach(() => { created = aborted = false; })
+
+      class AbortControllerMock {
+        constructor() {
+          created = true;
+        }
         signal: {};
         abort = () => {
-          called = true;
+          aborted = true;
         };
       }
 
-      (global as any).AbortController = AbortController;
-
-      fetch.mockReturnValueOnce(Promise.resolve({ text }));
-      text.mockReturnValueOnce(
-        Promise.resolve('{ "data": { "hello": "world" } }'),
-      );
-
-      const link = createHttpLink({ uri: 'data', fetch: fetch as any });
-
-      const sub = execute(link, { query: sampleQuery }).subscribe({
-        next: result => {
-          reject('result should not have been called');
+      const originalAbortController = globalThis.AbortController;
+      beforeAll(() => {
+        globalThis.AbortController = AbortControllerMock as any;
+      })
+      afterAll(() => {
+        globalThis.AbortController = originalAbortController;
+      })
+
+      beforeEach( () => {
+        fetch.mockResolvedValueOnce({ text });
+        text.mockResolvedValueOnce('{ "data": { "hello": "world" } }');
+      })
+      afterEach(() => {
+        fetch.mockReset();
+        text.mockReset();
+      })
+
+      const failingObserver: Observer<FetchResult> = {
+        next: () => {
+          fail('result should not have been called');
         },
         error: e => {
-          reject(e);
+          fail(e);
         },
         complete: () => {
-          reject('complete should not have been called');
+          fail('complete should not have been called');
         },
+      }
+
+      it('unsubscribing cancels internally created AbortController', () => {
+        const link = createHttpLink({ uri: 'data', fetch: fetch as any });
+
+        const sub = execute(link, { query: sampleQuery }).subscribe(failingObserver);
+        sub.unsubscribe();
+
+        expect(created).toBe(true);
+        expect(aborted).toBe(true);
       });
-      sub.unsubscribe();
 
-      setTimeout(
-        makeCallback(resolve, reject, () => {
-          delete (global as any).AbortController;
-          expect(called).toBe(true);
-          fetch.mockReset();
-          text.mockReset();
-        }),
-        150,
-      );
+      it('passing a signal in means no AbortController is created internally', () => {
+        const link = createHttpLink({ uri: 'data', fetch: fetch as any, fetchOptions: { signal: {} } });
+
+        const sub = execute(link, { query: sampleQuery } ).subscribe(failingObserver);
+        sub.unsubscribe();
+
+        expect(created).toBe(false);
+        expect(aborted).toBe(false);
+      });
+
+      it('resolving fetch does not cause the AbortController to be aborted', async () => {
+        // (the request is already finished at that point)
+        const link = createHttpLink({ uri: 'data', fetch: fetch as any });
+
+        await new Promise<void>(resolve => execute(link, { query: sampleQuery }).subscribe({
+          complete: resolve
+        }));
+
+        expect(created).toBe(true);
+        expect(aborted).toBe(false);
+      });
+
+      it('throwing an error from fetch does not cause the AbortController to be aborted', async () => {
+        fetch.mockReset();
+        fetch.mockRejectedValueOnce("This is an error!")
+        // the request would be closed by the browser in the case of an error anyways
+        const link = createHttpLink({ uri: 'data', fetch: fetch as any });
+
+        await new Promise<void>(resolve => execute(link, { query: sampleQuery }).subscribe({
+          error: resolve
+        }));
+
+        expect(created).toBe(true);
+        expect(aborted).toBe(false);
+      });
     });
 
     const body = '{';

src/link/http/createHttpLink.ts

@@ -9,7 +9,7 @@ import { selectURI } from './selectURI';
 import {
   handleError,
   readMultipartBody,
-  readJsonBody
+  parseAndCheckHttpResponse
 } from './parseAndCheckHttpResponse';
 import { checkFetcher } from './checkFetcher';
 import type {
@@ -20,7 +20,6 @@ import {
   defaultPrinter,
   fallbackHttpConfig
 } from './selectHttpOptionsAndBody';
-import { createSignalIfSupported } from './createSignalIfSupported';
 import { rewriteURIForGET } from './rewriteURIForGET';
 import { fromError, filterOperationVariables } from '../utils';
 import {
@@ -119,11 +118,10 @@ export const createHttpLink = (linkOptions: HttpOptions = {}) => {
       body.variables = filterOperationVariables(body.variables, operation.query);
     }
 
-    let controller: any;
-    if (!(options as any).signal) {
-      const { controller: _controller, signal } = createSignalIfSupported();
-      controller = _controller;
-      if (controller) (options as any).signal = signal;
+    let controller: AbortController | undefined;
+    if (!options.signal && typeof AbortController !== 'undefined') {
+      controller = new AbortController();
+      options.signal = controller.signal;
     }
 
     // If requested, set method to GET if there are no mutations.
@@ -182,18 +180,26 @@ export const createHttpLink = (linkOptions: HttpOptions = {}) => {
       // removal of window.fetch, which is unlikely but not impossible.
       const currentFetch = preferredFetch || maybe(() => fetch) || backupFetch;
 
+      const observerNext = observer.next.bind(observer);
       currentFetch!(chosenURI, options)
         .then(response => {
           operation.setContext({ response });
           const ctype = response.headers?.get('content-type');
 
           if (ctype !== null && /^multipart\/mixed/i.test(ctype)) {
-            return readMultipartBody(response, observer);
+            return readMultipartBody(response, observerNext);
           } else {
-            return readJsonBody(response, operation, observer);
+            return parseAndCheckHttpResponse(operation)(response).then(observerNext);
           }
         })
-        .catch(err => handleError(err, observer));
+        .then(() => {
+          controller = undefined;
+          observer.complete();
+        })
+        .catch(err => {
+          controller = undefined;
+          handleError(err, observer)
+        });
 
       return () => {
         // XXX support canceling this request

src/link/http/createSignalIfSupported.ts

@@ -1,3 +1,8 @@
+/**
+ * @deprecated 
+ * This is not used internally any more and will be removed in
+ * the next major version of Apollo Client.
+ */
 export const createSignalIfSupported = () => {
   if (typeof AbortController === 'undefined')
     return { controller: false, signal: false };

src/link/http/parseAndCheckHttpResponse.ts

@@ -17,7 +17,7 @@ export type ServerParseError = Error & {
 
 export async function readMultipartBody<
   T extends object = Record<string, unknown>
->(response: Response, observer: Observer<T>) {
+>(response: Response, nextValue: (value: T) => void) {
   if (TextDecoder === undefined) {
     throw new Error(
       "TextDecoder must be defined in the environment: please import a polyfill."
@@ -74,52 +74,47 @@ export async function readMultipartBody<
       const body = message.slice(i);
 
       if (body) {
-        try {
-          const result = parseJsonBody<T>(response, body);
-          if (
-            Object.keys(result).length > 1 ||
-            "data" in result ||
-            "incremental" in result ||
-            "errors" in result ||
-            "payload" in result
-          ) {
-            if (isApolloPayloadResult(result)) {
-              let next = {};
-              if ("payload" in result) {
-                next = { ...result.payload };
-              }
-              if ("errors" in result) {
-                next = {
-                  ...next,
-                  extensions: {
-                    ...("extensions" in next ? next.extensions : null as any),
-                    [PROTOCOL_ERRORS_SYMBOL]: result.errors
-                  },
-                };
-              }
-              observer.next?.(next as T);
-            } else {
-              // for the last chunk with only `hasNext: false`
-              // we don't need to call observer.next as there is no data/errors
-              observer.next?.(result);
+        const result = parseJsonBody<T>(response, body);
+        if (
+          Object.keys(result).length > 1 ||
+          "data" in result ||
+          "incremental" in result ||
+          "errors" in result ||
+          "payload" in result
+        ) {
+          if (isApolloPayloadResult(result)) {
+            let next = {};
+            if ("payload" in result) {
+              next = { ...result.payload };
+            }
+            if ("errors" in result) {
+              next = {
+                ...next,
+                extensions: {
+                  ...("extensions" in next ? next.extensions : null as any),
+                  [PROTOCOL_ERRORS_SYMBOL]: result.errors
+                },
+              };
             }
-          } else if (
-            // If the chunk contains only a "hasNext: false", we can call
-            // observer.complete() immediately.
-            Object.keys(result).length === 1 &&
-            "hasNext" in result &&
-            !result.hasNext
-          ) {
-            observer.complete?.();
+            nextValue(next as T);
+          } else {
+            // for the last chunk with only `hasNext: false`
+            // we don't need to call observer.next as there is no data/errors
+            nextValue(result);
           }
-        } catch (err) {
-          handleError(err, observer);
+        } else if (
+          // If the chunk contains only a "hasNext: false", we can call
+          // observer.complete() immediately.
+          Object.keys(result).length === 1 &&
+          "hasNext" in result &&
+          !result.hasNext
+        ) {
+          return;
         }
       }
       bi = buffer.indexOf(boundary);
     }
   }
-  observer.complete?.();
 }
 
 export function parseHeaders(headerText: string): Record<string, string> {
@@ -206,19 +201,6 @@ export function handleError(err: any, observer: Observer<any>) {
   observer.error?.(err);
 }
 
-export function readJsonBody<T = Record<string, unknown>>(
-  response: Response,
-  operation: Operation,
-  observer: Observer<T>
-) {
-  parseAndCheckHttpResponse(operation)(response)
-    .then((result) => {
-      observer.next?.(result);
-      observer.complete?.();
-    })
-    .catch((err) => handleError(err, observer));
-}
-
 export function parseAndCheckHttpResponse(operations: Operation | Operation[]) {
   return (response: Response) =>
     response