Prevent client prefetch stream from closing (#72420)

Based on:
- #72418 

---

When PPR is enabled, prefetch streams may contain references that never
resolve, because that's how we encode dynamic data access. In the
decoded object returned by the Flight client, these are reified into
hanging promises that suspend during render, which is effectively what
we want. The UI resolves when it switches to the dynamic data stream
(via useDeferredValue(dynamic, static)).

However, the Flight implementation currently errors if the server closes
the response before all the references are resolved. As a cheat to work
around this, we wrap the original stream in a new stream that never
closes, and therefore doesn't error.

Author: acdlite

packages/next/src/client/components/router-reducer/fetch-server-response.ts

@@ -213,8 +213,11 @@ export async function fetchServerResponse(
     }
 
     // Handle the `fetch` readable stream that can be unwrapped by `React.use`.
+    const flightStream = postponed
+      ? createUnclosingPrefetchStream(res.body)
+      : res.body
     const response: NavigationFlightResponse = await createFromReadableStream(
-      res.body,
+      flightStream,
       { callServer, findSourceMapURL }
     )
 
@@ -248,3 +251,36 @@ export async function fetchServerResponse(
     }
   }
 }
+
+function createUnclosingPrefetchStream(
+  originalFlightStream: ReadableStream<Uint8Array>
+): ReadableStream<Uint8Array> {
+  // When PPR is enabled, prefetch streams may contain references that never
+  // resolve, because that's how we encode dynamic data access. In the decoded
+  // object returned by the Flight client, these are reified into hanging
+  // promises that suspend during render, which is effectively what we want.
+  // The UI resolves when it switches to the dynamic data stream
+  // (via useDeferredValue(dynamic, static)).
+  //
+  // However, the Flight implementation currently errors if the server closes
+  // the response before all the references are resolved. As a cheat to work
+  // around this, we wrap the original stream in a new stream that never closes,
+  // and therefore doesn't error.
+  const reader = originalFlightStream.getReader()
+  return new ReadableStream({
+    async pull(controller) {
+      while (true) {
+        const { done, value } = await reader.read()
+        if (!done) {
+          // Pass to the target stream and keep consuming the Flight response
+          // from the server.
+          controller.enqueue(value)
+          continue
+        }
+        // The server stream has closed. Exit, but intentionally do not close
+        // the target stream.
+        return
+      }
+    },
+  })
+}