Fix no-cache stream/defer case where streaming was incorrectly reported

Author: jerelmiller

src/core/QueryInfo.ts

@@ -271,27 +271,35 @@ export class QueryInfo<
 
     let result: MarkQueryResult<any, ExtensionsWithStreamInfo> = {
       ...incrementalResult,
-      dataState:
-        incrementalResult.data == null ? "empty"
-          // TODO: This is too naive. For stream arrays, this might be complete
-        : this.hasNext ? "streaming"
-        : "complete",
+      dataState: incrementalResult.data == null ? "empty" : "complete",
     };
 
     if (skipCache) {
-      return {
-        ...result,
-        dataState:
-          result.data == null ? "empty"
-            // Ww can simplify the checks for no-cache queries because the
-            // result is purely server driven which means we can assume a
-            // well-formed GraphQL response. In this case, `streaming` only
-            // makes sense if we are using the `@defer` directive and there are
-            // more chunks still streaming (i.e. this.hasNext is true). Purely
-            // `@stream` queries should always be complete.
-          : this.hasNext && hasDirectives(["defer"], query) ? "streaming"
-          : "complete",
-      };
+      const hasPendingDefer = this.incremental?.pending?.some(
+        (pending) => this.incremental?.getPendingType?.(pending.id) === "defer"
+      );
+
+      if (
+        hasPendingDefer ||
+        // The Defer20220824Handler cannot track pending/completed incremental
+        // chunks due to its data format so we naively set dataState to
+        // streaming if we are still processing chunks. The only case where
+        // streaming is incorrect and should actually be complete is when
+        // both a @defer and @stream boundary is present and the @defer chunk
+        // has completed before the `@stream` array.
+        //
+        // Assigning the naive "streaming" value avoids a much more expensive
+        // pass over `result.data` that would otherwise need to traverse the
+        // selection sets and evaluate the data object at each defer boundary
+        // to see if it fulfills the selection set. For such a narrow case where
+        // its incorrect on a format that is now outdated is not worth the
+        // fix so we are ok with reporting a `streaming` here.
+        (!this.incremental?.pending && this.hasNext)
+      ) {
+        result.dataState = "streaming";
+      }
+
+      return result;
     }
 
     if (shouldWriteResult(result, errorPolicy)) {

src/core/__tests__/client.watchQuery/streamGraphQL17Alpha9.test.ts

@@ -896,6 +896,100 @@ test("handles @defer inside @stream", async () => {
   await expect(observableStream).not.toEmitAnything();
 });
 
+test('reports "complete" once a sibling @defer boundary resolves while a @stream is still streaming with a "no-cache" fetch policy', async () => {
+  const { httpLink, enqueueInitialChunk, enqueueSubsequentChunk } =
+    mockDeferStreamGraphQL17Alpha9();
+  const client = new ApolloClient({
+    link: httpLink,
+    cache: new InMemoryCache(),
+    incrementalHandler: new GraphQL17Alpha9Handler(),
+  });
+
+  const query = gql`
+    query {
+      friendList @stream(initialCount: 1) {
+        id
+        name
+      }
+      ... @defer {
+        someScalar
+      }
+    }
+  `;
+
+  const observableStream = new ObservableStream(
+    client.watchQuery({ query, fetchPolicy: "no-cache" })
+  );
+
+  await expect(observableStream).toEmitTypedValue({
+    data: undefined,
+    dataState: "empty",
+    loading: true,
+    networkStatus: NetworkStatus.loading,
+    partial: true,
+  });
+
+  enqueueInitialChunk({
+    data: { friendList: [{ __typename: "Friend", id: "1", name: "Luke" }] },
+    pending: [
+      { id: "0", path: [] },
+      { id: "1", path: ["friendList"] },
+    ],
+    hasNext: true,
+  } as any);
+
+  await expect(observableStream).toEmitTypedValue({
+    data: markAsStreaming({
+      friendList: [{ __typename: "Friend", id: "1", name: "Luke" }],
+    }),
+    dataState: "streaming",
+    loading: true,
+    networkStatus: NetworkStatus.streaming,
+    partial: true,
+  });
+
+  enqueueSubsequentChunk({
+    incremental: [{ id: "0", data: { someScalar: "resolved" } }],
+    completed: [{ id: "0" }],
+    hasNext: true,
+  } as any);
+
+  await expect(observableStream).toEmitTypedValue({
+    data: {
+      friendList: [{ __typename: "Friend", id: "1", name: "Luke" }],
+      someScalar: "resolved",
+    },
+    dataState: "complete",
+    loading: true,
+    networkStatus: NetworkStatus.streaming,
+    partial: false,
+  });
+
+  enqueueSubsequentChunk({
+    incremental: [
+      { id: "1", items: [{ __typename: "Friend", id: "2", name: "Han" }] },
+    ],
+    completed: [{ id: "1" }],
+    hasNext: false,
+  } as any);
+
+  await expect(observableStream).toEmitTypedValue({
+    data: {
+      friendList: [
+        { __typename: "Friend", id: "1", name: "Luke" },
+        { __typename: "Friend", id: "2", name: "Han" },
+      ],
+      someScalar: "resolved",
+    },
+    dataState: "complete",
+    loading: false,
+    networkStatus: NetworkStatus.ready,
+    partial: false,
+  });
+
+  await expect(observableStream).not.toEmitAnything();
+});
+
 test("can use custom merge function to combine cached and streamed lists", async () => {
   const cache = new InMemoryCache({
     typePolicies: {