fix potential memory leak in Concast, add tests (#11358)

.changeset/forty-cups-shop.md

@@ -0,0 +1,5 @@
+---
+"@apollo/client": patch
+---
+
+Fixes a potential memory leak in `Concast` that might have been triggered when `Concast` was used outside of Apollo Client.

.size-limits.json

@@ -1,4 +1,4 @@
 {
-  "dist/apollo-client.min.cjs": 38630,
-  "import { ApolloClient, InMemoryCache, HttpLink } from \"dist/index.js\" (production)": 32213
+  "dist/apollo-client.min.cjs": 38632,
+  "import { ApolloClient, InMemoryCache, HttpLink } from \"dist/index.js\" (production)": 32215
 }

package.json

@@ -48,7 +48,7 @@
     "ci:precheck": "node config/precheck.js",
     "format": "prettier --write .",
     "lint": "eslint 'src/**/*.{[jt]s,[jt]sx}'",
-    "test": "jest --config ./config/jest.config.js",
+    "test": "node --expose-gc ./node_modules/jest/bin/jest.js --config ./config/jest.config.js",
     "test:debug": "node --inspect-brk node_modules/.bin/jest --config ./config/jest.config.js --runInBand --testTimeout 99999 --logHeapUsage",
     "test:ci": "TEST_ENV=ci npm run test:coverage -- --logHeapUsage && npm run test:memory",
     "test:watch": "jest --config ./config/jest.config.js --watch",

src/testing/matchers/index.d.ts

@@ -43,6 +43,10 @@ interface ApolloCustomMatchers<R = void, T = {}> {
     | ProfiledHook<any, any>
     ? (count: number, options?: NextRenderOptions) => Promise<R>
     : { error: "matcher needs to be called on a ProfiledComponent instance" };
+
+  toBeGarbageCollected: T extends WeakRef<any>
+    ? () => Promise<R>
+    : { error: "matcher needs to be called on a WeakRef instance" };
 }
 
 declare global {

src/testing/matchers/index.ts

@@ -2,10 +2,12 @@ import { expect } from "@jest/globals";
 import { toMatchDocument } from "./toMatchDocument.js";
 import { toHaveSuspenseCacheEntryUsing } from "./toHaveSuspenseCacheEntryUsing.js";
 import { toRerender, toRenderExactlyTimes } from "./ProfiledComponent.js";
+import { toBeGarbageCollected } from "./toBeGarbageCollected.js";
 
 expect.extend({
   toHaveSuspenseCacheEntryUsing,
   toMatchDocument,
   toRerender,
   toRenderExactlyTimes,
+  toBeGarbageCollected,
 });

src/testing/matchers/toBeGarbageCollected.ts

@@ -0,0 +1,59 @@
+import type { MatcherFunction } from "expect";
+
+// this is necessary because this file is picked up by `tsc` (it's not a test),
+// but our main `tsconfig.json` doesn't include `"ES2021.WeakRef"` on purpose
+declare class WeakRef<T extends WeakKey> {
+  constructor(target: T);
+  deref(): T | undefined;
+}
+
+export const toBeGarbageCollected: MatcherFunction<[weakRef: WeakRef<any>]> =
+  async function (actual) {
+    const hint = this.utils.matcherHint("toBeGarbageCollected");
+
+    if (!(actual instanceof WeakRef)) {
+      throw new Error(
+        hint +
+          "\n\n" +
+          `Expected value to be a WeakRef, but it was a ${typeof actual}.`
+      );
+    }
+
+    let pass = false;
+    let interval: NodeJS.Timeout | undefined;
+    let timeout: NodeJS.Timeout | undefined;
+    await Promise.race([
+      new Promise<void>((resolve) => {
+        timeout = setTimeout(resolve, 1000);
+      }),
+      new Promise<void>((resolve) => {
+        interval = setInterval(() => {
+          global.gc!();
+          pass = actual.deref() === undefined;
+          if (pass) {
+            resolve();
+          }
+        }, 1);
+      }),
+    ]);
+
+    clearInterval(interval);
+    clearTimeout(timeout);
+
+    return {
+      pass,
+      message: () => {
+        if (pass) {
+          return (
+            hint +
+            "\n\n" +
+            "Expected value to not be cache-collected, but it was."
+          );
+        }
+
+        return (
+          hint + "\n\n Expected value to be cache-collected, but it was not."
+        );
+      },
+    };
+  };

src/tsconfig.json

@@ -5,7 +5,7 @@
 {
   "compilerOptions": {
     "noEmit": true,
-    "lib": ["es2015", "esnext.asynciterable", "dom"],
+    "lib": ["es2015", "esnext.asynciterable", "dom", "ES2021.WeakRef"],
     "types": ["jest", "node", "./testing/matchers/index.d.ts"]
   },
   "extends": "../tsconfig.json",

src/utilities/observables/Concast.ts

@@ -210,7 +210,10 @@ export class Concast<T> extends Observable<T> {
           // followed by a 'complete' message (see addObserver).
           iterateObserversSafely(this.observers, "complete");
         } else if (isPromiseLike(value)) {
-          value.then((obs) => (this.sub = obs.subscribe(this.handlers)));
+          value.then(
+            (obs) => (this.sub = obs.subscribe(this.handlers)),
+            this.handlers.error
+          );
         } else {
           this.sub = value.subscribe(this.handlers);
         }

src/utilities/observables/__tests__/Concast.ts

@@ -1,5 +1,5 @@
 import { itAsync } from "../../../testing/core";
-import { Observable } from "../Observable";
+import { Observable, Observer } from "../Observable";
 import { Concast, ConcastSourcesIterable } from "../Concast";
 
 describe("Concast Observable (similar to Behavior Subject in RxJS)", () => {
@@ -187,4 +187,115 @@ describe("Concast Observable (similar to Behavior Subject in RxJS)", () => {
       sub.unsubscribe();
     });
   });
+
+  it("resolving all sources of a concast frees all observer references on `this.observers`", async () => {
+    const { promise, resolve } = deferred<Observable<number>>();
+    const observers: Observer<any>[] = [{ next() {} }];
+    const observerRefs = observers.map((observer) => new WeakRef(observer));
+
+    const concast = new Concast<number>([Observable.of(1, 2), promise]);
+
+    concast.subscribe(observers[0]);
+    delete observers[0];
+
+    expect(concast["observers"].size).toBe(1);
+
+    resolve(Observable.of(3, 4));
+
+    await expect(concast.promise).resolves.toBe(4);
+
+    await expect(observerRefs[0]).toBeGarbageCollected();
+  });
+
+  it("rejecting a source-wrapping promise of a concast frees all observer references on `this.observers`", async () => {
+    const { promise, reject } = deferred<Observable<number>>();
+    let subscribingObserver: Observer<any> | undefined = {
+      next() {},
+      error() {},
+    };
+    const subscribingObserverRef = new WeakRef(subscribingObserver);
+
+    const concast = new Concast<number>([
+      Observable.of(1, 2),
+      promise,
+      // just to ensure this also works if the cancelling source is not the last source
+      Observable.of(3, 5),
+    ]);
+
+    concast.subscribe(subscribingObserver);
+
+    expect(concast["observers"].size).toBe(1);
+
+    reject("error");
+    await expect(concast.promise).rejects.toBe("error");
+    subscribingObserver = undefined;
+    await expect(subscribingObserverRef).toBeGarbageCollected();
+  });
+
+  it("rejecting a source of a concast frees all observer references on `this.observers`", async () => {
+    let subscribingObserver: Observer<any> | undefined = {
+      next() {},
+      error() {},
+    };
+    const subscribingObserverRef = new WeakRef(subscribingObserver);
+
+    let sourceObserver!: Observer<number>;
+    const sourceObservable = new Observable<number>((o) => {
+      sourceObserver = o;
+    });
+
+    const concast = new Concast<number>([
+      Observable.of(1, 2),
+      sourceObservable,
+      Observable.of(3, 5),
+    ]);
+
+    concast.subscribe(subscribingObserver);
+
+    expect(concast["observers"].size).toBe(1);
+
+    await Promise.resolve();
+    sourceObserver.error!("error");
+    await expect(concast.promise).rejects.toBe("error");
+    subscribingObserver = undefined;
+    await expect(subscribingObserverRef).toBeGarbageCollected();
+  });
+
+  it("after subscribing to an already-resolved concast, the reference is freed up again", async () => {
+    const concast = new Concast<number>([Observable.of(1, 2)]);
+    await expect(concast.promise).resolves.toBe(2);
+    await Promise.resolve();
+
+    let sourceObserver: Observer<any> | undefined = { next() {}, error() {} };
+    const sourceObserverRef = new WeakRef(sourceObserver);
+
+    concast.subscribe(sourceObserver);
+
+    sourceObserver = undefined;
+    await expect(sourceObserverRef).toBeGarbageCollected();
+  });
+
+  it("after subscribing to an already-rejected concast, the reference is freed up again", async () => {
+    const concast = new Concast<number>([Promise.reject("error")]);
+    await expect(concast.promise).rejects.toBe("error");
+    await Promise.resolve();
+
+    let sourceObserver: Observer<any> | undefined = { next() {}, error() {} };
+    const sourceObserverRef = new WeakRef(sourceObserver);
+
+    concast.subscribe(sourceObserver);
+
+    sourceObserver = undefined;
+    await expect(sourceObserverRef).toBeGarbageCollected();
+  });
 });
+
+function deferred<X>() {
+  let resolve!: (v: X) => void;
+  let reject!: (e: any) => void;
+  const promise = new Promise<X>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { resolve, reject, promise };
+}