close on unsubscribe, catch errors

integration-test/nextjs/src/app/rsc/dynamic/PreloadQuery/error.tsx

@@ -0,0 +1,17 @@
+"use client"; // Error boundaries must be Client Components
+
+export default function Error({
+  error,
+  reset,
+}: {
+  error: Error & { digest?: string };
+  reset: () => void;
+}) {
+  return (
+    <>
+      <h2>Encountered an error:</h2>
+      <pre>{error.message}</pre>
+      <button onClick={() => reset()}>Try again</button>
+    </>
+  );
+}

integration-test/nextjs/src/app/rsc/dynamic/PreloadQuery/queryRef-useReadQuery/ClientChild.tsx

@@ -2,7 +2,7 @@
 
 import { useQueryRefHandlers, useReadQuery } from "@apollo/client";
 import { DynamicProductResult } from "../shared";
-import { TransportedQueryRef } from "@apollo/experimental-nextjs-app-support/ssr";
+import { TransportedQueryRef } from "@apollo/experimental-nextjs-app-support";
 
 export function ClientChild({
   queryRef,

packages/client-react-streaming/src/ReadableStreamLink.tsx

@@ -75,6 +75,7 @@ export const TeeToReadableStreamLink = new ApolloLink((operation, forward) => {
       );
 
       return () => {
+        controller.close();
         subscription.unsubscribe();
       };
     });

packages/client-react-streaming/src/transportedQueryRef.ts

@@ -114,19 +114,23 @@ export function createTransportedQueryPreloader(
 
     // Instead of creating the queryRef, we kick off a query that will feed the network response
     // into our custom event stream.
-    client.query({
-      query,
-      ...options,
-      // ensure that this query makes it to the network
-      fetchPolicy: "network-only",
-      context: skipDataTransport(
-        teeToReadableStream(__injectIntoStream!, {
-          ...options?.context,
-          // we want to do this even if the query is already running for another reason
-          queryDeduplication: false,
-        })
-      ),
-    });
+    client
+      .query({
+        query,
+        ...options,
+        // ensure that this query makes it to the network
+        fetchPolicy: "network-only",
+        context: skipDataTransport(
+          teeToReadableStream(__injectIntoStream!, {
+            ...options?.context,
+            // we want to do this even if the query is already running for another reason
+            queryDeduplication: false,
+          })
+        ),
+      })
+      .catch(() => {
+        /* we want to avoid any floating promise rejections */
+      });
 
     return createTransportedQueryRef<any, any>(
       query,