ObservableQuery: only report results for the current variables (#11078)

* Add reproduction

* ObservableQuery: only report results for the current variables

* changeset

* chores

* add tests directly to `ObservableStream`

* adjust fetchPolicy

* add failing tests for missing cache write

* remove unrelated test part

* Update smooth-plums-shout.md

* Clean up Prettier, Size-limit, and Api-Extractor

* review suggestions

---------

Co-authored-by: Jan Amann <jan@amann.work>
Co-authored-by: phryneas <phryneas@users.noreply.github.com>

.changeset/smooth-plums-shout.md

@@ -0,0 +1,5 @@
+---
+"@apollo/client": patch
+---
+
+ObservableQuery: prevent reporting results of previous queries if the variables changed since

.size-limits.json

@@ -1,4 +1,4 @@
 {
-  "dist/apollo-client.min.cjs": 39130,
-  "import { ApolloClient, InMemoryCache, HttpLink } from \"dist/index.js\" (production)": 32651
+  "dist/apollo-client.min.cjs": 39136,
+  "import { ApolloClient, InMemoryCache, HttpLink } from \"dist/index.js\" (production)": 32663
 }

src/__tests__/mutationResults.ts

@@ -1186,8 +1186,6 @@ describe("mutation results", () => {
 
       subscribeAndCount(reject, watchedQuery, (count, result) => {
         if (count === 1) {
-          expect(result.data).toEqual({ echo: "a" });
-        } else if (count === 2) {
           expect(result.data).toEqual({ echo: "b" });
           client.mutate({
             mutation: resetMutation,
@@ -1197,7 +1195,7 @@ describe("mutation results", () => {
               },
             },
           });
-        } else if (count === 3) {
+        } else if (count === 2) {
           expect(result.data).toEqual({ echo: "0" });
           resolve();
         }

src/core/ObservableQuery.ts

@@ -906,12 +906,16 @@ Did you mean to call refetch(variables) instead of refetch({ variables })?`,
     const { concast, fromLink } = this.fetch(options, newNetworkStatus, query);
     const observer: Observer<ApolloQueryResult<TData>> = {
       next: (result) => {
-        finishWaitingForOwnResult();
-        this.reportResult(result, variables);
+        if (equal(this.variables, variables)) {
+          finishWaitingForOwnResult();
+          this.reportResult(result, variables);
+        }
       },
       error: (error) => {
-        finishWaitingForOwnResult();
-        this.reportError(error, variables);
+        if (equal(this.variables, variables)) {
+          finishWaitingForOwnResult();
+          this.reportError(error, variables);
+        }
       },
     };
 

src/core/__tests__/ObservableQuery.ts

@@ -34,6 +34,7 @@ import wrap from "../../testing/core/wrap";
 import { resetStore } from "./QueryManager";
 import { SubscriptionObserver } from "zen-observable-ts";
 import { waitFor } from "@testing-library/react";
+import { ObservableStream } from "../../testing/internal";
 
 export const mockFetchQuery = (queryManager: QueryManager<any>) => {
   const fetchConcastWithInfo = queryManager["fetchConcastWithInfo"];
@@ -1086,6 +1087,98 @@ describe("ObservableQuery", () => {
       }
     );
 
+    it("calling refetch with different variables before the query itself resolved will only yield the result for the new variables", async () => {
+      const observers: SubscriptionObserver<FetchResult<typeof dataOne>>[] = [];
+      const queryManager = new QueryManager({
+        cache: new InMemoryCache(),
+        link: new ApolloLink((operation, forward) => {
+          return new Observable((observer) => {
+            observers.push(observer);
+          });
+        }),
+      });
+      const observableQuery = queryManager.watchQuery({
+        query,
+        variables: { id: 1 },
+      });
+      const stream = new ObservableStream(observableQuery);
+
+      observableQuery.refetch({ id: 2 });
+
+      observers[0].next({ data: dataOne });
+      observers[0].complete();
+
+      observers[1].next({ data: dataTwo });
+      observers[1].complete();
+
+      {
+        const result = await stream.takeNext();
+        expect(result).toEqual({
+          loading: false,
+          networkStatus: NetworkStatus.ready,
+          data: dataTwo,
+        });
+      }
+      expect(stream.take()).rejects.toThrow(/Timeout/i);
+    });
+
+    it("calling refetch multiple times with different variables will return only results for the most recent variables", async () => {
+      const observers: SubscriptionObserver<FetchResult<typeof dataOne>>[] = [];
+      const queryManager = new QueryManager({
+        cache: new InMemoryCache(),
+        link: new ApolloLink((operation, forward) => {
+          return new Observable((observer) => {
+            observers.push(observer);
+          });
+        }),
+      });
+      const observableQuery = queryManager.watchQuery({
+        query,
+        variables: { id: 1 },
+      });
+      const stream = new ObservableStream(observableQuery);
+
+      observers[0].next({ data: dataOne });
+      observers[0].complete();
+
+      {
+        const result = await stream.takeNext();
+        expect(result).toEqual({
+          loading: false,
+          networkStatus: NetworkStatus.ready,
+          data: dataOne,
+        });
+      }
+
+      observableQuery.refetch({ id: 2 });
+      observableQuery.refetch({ id: 3 });
+
+      observers[1].next({ data: dataTwo });
+      observers[1].complete();
+
+      observers[2].next({
+        data: {
+          people_one: {
+            name: "SomeOneElse",
+          },
+        },
+      });
+      observers[2].complete();
+
+      {
+        const result = await stream.takeNext();
+        expect(result).toEqual({
+          loading: false,
+          networkStatus: NetworkStatus.ready,
+          data: {
+            people_one: {
+              name: "SomeOneElse",
+            },
+          },
+        });
+      }
+    });
+
     itAsync(
       "calls fetchRequest with fetchPolicy `no-cache` when using `no-cache` fetch policy",
       (resolve, reject) => {

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

@@ -1,4 +1,4 @@
-import React, { Fragment, ReactNode, useEffect, useState } from "react";
+import React, { Fragment, ReactNode, useEffect, useRef, useState } from "react";
 import { DocumentNode, GraphQLError } from "graphql";
 import gql from "graphql-tag";
 import { act } from "react-dom/test-utils";
@@ -27,6 +27,7 @@ import { QueryResult } from "../../types/types";
 import { useQuery } from "../useQuery";
 import { useMutation } from "../useMutation";
 import { profileHook, spyOnConsole } from "../../../testing/internal";
+import { useApolloClient } from "../useApolloClient";
 
 describe("useQuery Hook", () => {
   describe("General use", () => {
@@ -4494,6 +4495,138 @@ describe("useQuery Hook", () => {
         });
       });
     });
+
+    it("keeps cache consistency when a call to refetchQueries is interrupted with another query caused by changing variables and the second query returns before the first one", async () => {
+      const CAR_QUERY_BY_ID = gql`
+        query Car($id: Int) {
+          car(id: $id) {
+            make
+            model
+          }
+        }
+      `;
+
+      const mocks = {
+        1: [
+          {
+            car: {
+              make: "Audi",
+              model: "A4",
+              __typename: "Car",
+            },
+          },
+          {
+            car: {
+              make: "Audi",
+              model: "A3", // Changed
+              __typename: "Car",
+            },
+          },
+        ],
+        2: [
+          {
+            car: {
+              make: "Audi",
+              model: "RS8",
+              __typename: "Car",
+            },
+          },
+        ],
+      };
+
+      const link = new ApolloLink(
+        (operation) =>
+          new Observable((observer) => {
+            if (operation.variables.id === 1) {
+              // Queries for this ID return after a delay
+              setTimeout(() => {
+                const data = mocks[1].splice(0, 1).pop();
+                observer.next({ data });
+                observer.complete();
+              }, 100);
+            } else if (operation.variables.id === 2) {
+              // Queries for this ID return immediately
+              const data = mocks[2].splice(0, 1).pop();
+              observer.next({ data });
+              observer.complete();
+            } else {
+              observer.error(new Error("Unexpected query"));
+            }
+          })
+      );
+
+      const hookResponse = jest.fn().mockReturnValue(null);
+
+      function Component({ children, id }: any) {
+        const result = useQuery(CAR_QUERY_BY_ID, {
+          variables: { id },
+          notifyOnNetworkStatusChange: true,
+          fetchPolicy: "network-only",
+        });
+        const client = useApolloClient();
+        const hasRefetchedRef = useRef(false);
+
+        useEffect(() => {
+          if (
+            result.networkStatus === NetworkStatus.ready &&
+            !hasRefetchedRef.current
+          ) {
+            client.reFetchObservableQueries();
+            hasRefetchedRef.current = true;
+          }
+        }, [result.networkStatus]);
+
+        return children(result);
+      }
+
+      const { rerender } = render(
+        <Component id={1}>{hookResponse}</Component>,
+        {
+          wrapper: ({ children }) => (
+            <MockedProvider link={link}>{children}</MockedProvider>
+          ),
+        }
+      );
+
+      await waitFor(() => {
+        // Resolves as soon as reFetchObservableQueries is
+        // called, but before the result is returned
+        expect(hookResponse).toHaveBeenCalledTimes(3);
+      });
+
+      rerender(<Component id={2}>{hookResponse}</Component>);
+
+      await waitFor(() => {
+        // All results are returned
+        expect(hookResponse).toHaveBeenCalledTimes(5);
+      });
+
+      expect(hookResponse.mock.calls.map((call) => call[0].data)).toEqual([
+        undefined,
+        {
+          car: {
+            __typename: "Car",
+            make: "Audi",
+            model: "A4",
+          },
+        },
+        {
+          car: {
+            __typename: "Car",
+            make: "Audi",
+            model: "A4",
+          },
+        },
+        undefined,
+        {
+          car: {
+            __typename: "Car",
+            make: "Audi",
+            model: "RS8",
+          },
+        },
+      ]);
+    });
   });
 
   describe("Callbacks", () => {

src/testing/internal/ObservableStream.ts

@@ -29,6 +29,7 @@ async function* observableToAsyncEventIterator<T>(observable: Observable<T>) {
     (error) => resolveNext({ type: "error", error }),
     () => resolveNext({ type: "complete" })
   );
+  yield "initialization value" as unknown as Promise<ObservableEvent<T>>;
 
   while (true) {
     yield promises.shift()!;
@@ -54,7 +55,11 @@ class IteratorStream<T> {
 
 export class ObservableStream<T> extends IteratorStream<ObservableEvent<T>> {
   constructor(observable: Observable<T>) {
-    super(observableToAsyncEventIterator(observable));
+    const iterator = observableToAsyncEventIterator(observable);
+    // we need to call next() once to start the generator so we immediately subscribe.
+    // the first value is always "initialization value" which we don't care about
+    iterator.next();
+    super(iterator);
   }
 
   async takeNext(options?: TakeOptions): Promise<T> {