Guarantee Concast cleanup without Observable cancelled prematurely rejection

CHANGELOG.md

@@ -1,5 +1,10 @@
 ## Apollo Client 3.6.4 (unreleased)
 
+### Bug Fixes
+
+- Guarantee `Concast` cleanup without `Observable cancelled prematurely` rejection, potentially solving long-standing issues involving that error. <br/>
+  [@benjamn](https://github.com/benjamn) in [#9701](https://github.com/apollographql/apollo-client/pull/9701)
+
 ### Improvements
 
 - Internalize `useSyncExternalStore` shim, for more control than `use-sync-external-store` provides, fixing some React Native issues. <br/>

src/utilities/observables/Concast.ts

@@ -141,10 +141,10 @@ export class Concast<T> extends Observable<T> {
     if (this.observers.delete(observer) &&
         --this.addCount < 1 &&
         !quietly) {
-      // In case there are still any cleanup observers in this.observers,
-      // and no error or completion has been broadcast yet, make sure
-      // those observers receive an error that terminates them.
-      this.handlers.error(new Error("Observable cancelled prematurely"));
+      // In case there are still any cleanup observers in this.observers, and no
+      // error or completion has been broadcast yet, make sure those observers
+      // have a chance to run and then remove themselves from this.observers.
+      this.handlers.complete();
     }
   }
 
@@ -187,9 +187,11 @@ export class Concast<T> extends Observable<T> {
     },
 
     complete: () => {
-      if (this.sub !== null) {
+      const { sub } = this;
+      if (sub !== null) {
         const value = this.sources.shift();
         if (!value) {
+          if (sub) setTimeout(() => sub.unsubscribe());
           this.sub = null;
           if (this.latest &&
               this.latest[0] === "next") {

src/utilities/observables/__tests__/Concast.ts

@@ -0,0 +1,78 @@
+import { itAsync } from "../../../testing/core";
+import { Observable } from "../Observable";
+import { Concast } from "../Concast";
+
+describe("Concast Observable (similar to Behavior Subject in RxJS)", () => {
+  itAsync("can concatenate other observables", (resolve, reject) => {
+    const concast = new Concast([
+      Observable.of(1, 2, 3),
+      Promise.resolve(Observable.of(4, 5)),
+      Observable.of(6, 7, 8, 9),
+      Promise.resolve().then(() => Observable.of(10)),
+      Observable.of(11),
+    ]);
+
+    const results: number[] = [];
+    concast.subscribe({
+      next(num) {
+        results.push(num);
+      },
+
+      error: reject,
+
+      complete() {
+        concast.promise.then(finalResult => {
+          expect(results).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
+          expect(finalResult).toBe(11);
+          resolve();
+        }).catch(reject);
+      },
+    });
+  });
+
+  itAsync("behaves appropriately if unsubscribed before first result", (resolve, reject) => {
+    const concast = new Concast([
+      new Promise(resolve => setTimeout(resolve, 100)).then(
+        () => Observable.of(1, 2, 3),
+      ),
+    ]);
+
+    const cleanupCounts = {
+      first: 0,
+      second: 0,
+    };
+
+    concast.cleanup(() => {
+      ++cleanupCounts.first;
+    });
+
+    const unsubscribe = concast.subscribe({
+      next() {
+        reject("should not have called observer.next");
+      },
+      error() {
+        reject("should not have called observer.error");
+      },
+      complete() {
+        reject("should not have called observer.complete");
+      },
+    });
+
+    concast.cleanup(() => {
+      ++cleanupCounts.second;
+    });
+
+    // Immediately unsubscribe the observer we just added, triggering
+    // completion.
+    unsubscribe.unsubscribe();
+
+    return concast.promise.then(finalResult => {
+      expect(finalResult).toBeUndefined();
+      expect(cleanupCounts).toEqual({
+        first: 1,
+        second: 1,
+      });
+      resolve();
+    }).catch(reject);
+  });
+});