Shift scope of Concast unsubscription fix from Concast to useSubscrip…

…tion

.changeset/sour-meals-compete.md

@@ -2,4 +2,4 @@
 "@apollo/client": patch
 ---
 
-Improve consistency when unsubscribing Concast with rollback when new observer subscribes to otherwise-active Concast right after last observer unsubscribes
+Delay Concast subscription teardown slightly in `useSubscription` to prevent unexpected Concast closure when one `useSubscription` hook tears down its in-flight Concast subscription immediately followed by another `useSubscription` hook reusing and subscribing to that same Concast

src/react/hooks/__tests__/useSubscription.test.tsx

@@ -996,4 +996,73 @@ describe('useSubscription Hook', () => {
       );
     });
   });
+
+  it('should handle simple subscription after old in-flight teardown immediately \
+followed by new in-flight setup', async () => {
+    const subscription = gql`
+      subscription {
+        car {
+          make
+        }
+      }
+    `;
+
+    const results = ['Audi', 'BMW'].map(make => ({
+      result: { data: { car: { make } } },
+    }));
+
+    const link = new MockSubscriptionLink();
+    const client = new ApolloClient({
+      link,
+      cache: new Cache({ addTypename: false })
+    });
+
+    const { result, unmount, rerender } = renderHook(
+      ({ coin }) => {
+        const heads = useSubscription(subscription, {
+          variables: {},
+          skip: coin === 'tails',
+          context: { coin: 'heads' }
+        });
+        const tails = useSubscription(subscription, {
+          variables: {},
+          skip: coin === 'heads',
+          context: { coin: 'tails' }
+        });
+        return { heads, tails };
+      },
+      {
+        initialProps: {
+          coin: 'heads'
+        },
+        wrapper: ({ children }) => (
+          <ApolloProvider client={client}>
+            {children}
+          </ApolloProvider>
+        )
+      },
+    );
+
+    rerender({ coin: 'tails' });
+
+    await new Promise(resolve => setTimeout(() => resolve('wait'), 20));
+
+    link.simulateResult(results[0]);
+
+    await waitFor(() => {
+      expect(result.current.tails.data).toEqual(results[0].result.data);
+    }, { interval: 1 });
+    expect(result.current.heads.data).toBeUndefined();
+
+    rerender({ coin: 'heads' });
+
+    link.simulateResult(results[1]);
+
+    await waitFor(() => {
+      expect(result.current.heads.data).toEqual(results[1].result.data);
+    }, { interval: 1 });
+    expect(result.current.tails.data).toBeUndefined();
+
+    unmount();
+  });
 });

src/react/hooks/useSubscription.ts

@@ -160,7 +160,18 @@ export function useSubscription<TData = any, TVariables extends OperationVariabl
     });
 
     return () => {
+      // immediately stop receiving subscription values, but do not unsubscribe
+      // until after a short delay in case another useSubscription hook is
+      // reusing the same underlying observable and is about to subscribe
+      const ignoredSubscription = observable.subscribe({
+        next() {},
+        error() {},
+        complete() {},
+      });
       subscription.unsubscribe();
+      setTimeout(() => {
+        ignoredSubscription.unsubscribe();
+      });
     };
   }, [observable]);
 

src/utilities/observables/Concast.ts

@@ -152,40 +152,15 @@ export class Concast<T> extends Observable<T> {
   // easy way to observe the final state of the Concast.
   private resolve: (result?: T | PromiseLike<T>) => void;
   private reject: (reason: any) => void;
-  private internalPromise = new Promise<T>((resolve, reject) => {
+  public readonly promise = new Promise<T>((resolve, reject) => {
     this.resolve = resolve;
     this.reject = reject;
   });
-  public get promise() {
-    return this.internalPromise;
-  }
 
   // Name and argument of the most recently invoked observer method, used
   // to deliver latest results immediately to new observers.
   private latest?: ["next", T] | ["error", any];
 
-  // Delay unsubscribing from the underlying subscription slightly,
-  // so that immediately subscribing another observer can keep the
-  // subscription active.
-  private deferredUnsubscribe() {
-    const { sub } = this;
-    this.sub = null;
-    const isError = this.latest?.[0] === 'error';
-    if (sub) {
-      setTimeout(() => {
-        if (this.observers.size > 0 && !sub.closed && !isError) {
-          this.sub = sub;
-          this.internalPromise = new Promise<T>((resolve, reject) => {
-            this.resolve = resolve;
-            this.reject = reject;
-          });
-        } else {
-          sub.unsubscribe();
-        }
-      });
-    }
-  }
-
   // Bound handler functions that can be reused for every internal
   // subscription.
   private handlers = {
@@ -200,7 +175,11 @@ export class Concast<T> extends Observable<T> {
     error: (error: any) => {
       const { sub } = this;
       if (sub !== null) {
-        this.deferredUnsubscribe();
+        // Delay unsubscribing from the underlying subscription slightly,
+        // so that immediately subscribing another observer can keep the
+        // subscription active.
+        if (sub) setTimeout(() => sub.unsubscribe());
+        this.sub = null;
         this.latest = ["error", error];
         this.reject(error);
         this.notify("error", error);
@@ -218,7 +197,8 @@ export class Concast<T> extends Observable<T> {
         // eventually have been initialized to a non-empty array.
         const value = sources.shift();
         if (!value) {
-          this.deferredUnsubscribe();
+          if (sub) setTimeout(() => sub.unsubscribe());
+          this.sub = null;
           if (this.latest &&
               this.latest[0] === "next") {
             this.resolve(this.latest[1]);

src/utilities/observables/__tests__/Concast.ts

@@ -172,63 +172,4 @@ describe("Concast Observable (similar to Behavior Subject in RxJS)", () => {
       sub.unsubscribe();
     });
   });
-
-  itAsync("maintains subscription and promise when replacing last observer", (resolve, reject) => {
-    const concast = new Concast<number>([
-      new Observable<number>(observer => {
-        const timerIds = [setTimeout(() => {
-          observer.next(1);
-          timerIds.push(setTimeout(() => {
-            observer.next(2);
-            observer.complete();
-          }, 100))
-        }, 100)];
-
-        return () => {
-          for (const timerId of timerIds) {
-            clearTimeout(timerId);
-          }
-        };
-      })
-    ]);
-    const observer = {
-      next() {
-        reject(new Error("should not have called observer.next"));
-      },
-      error: reject,
-      complete() {
-        reject(new Error("should not have called observer.complete"));
-      },
-    };
-
-    concast.addObserver(observer);
-    concast.removeObserver(observer);
-
-    const secondObserver = {
-      ...observer,
-      next(value: number) {
-        if (value === 1) {
-          Promise.race([
-            new Promise(resolve => setTimeout(() => resolve(true))),
-            concast.promise.then(() => false, () => false)
-          ]).then(isPending => {
-            if (!isPending) {
-              reject("should not have resolved or rejected concast.promise by first observer.next");
-            }
-          });
-        }
-      },
-      complete() {
-        resolve();
-      }
-    };
-    concast.addObserver(secondObserver);
-
-    return new Promise(resolve => setTimeout(resolve, 1000)).then(() => {
-      if (!(concast as unknown as Record<string, unknown>)['sub']) {
-        reject(new Error("should not have nulled out concast.sub yet"));
-      }
-      reject(new Error("unexpectedly timed out"));
-    });
-  });
 });