re-enable pagesToBench for stats app

packages/next/src/server/app-render/app-render.tsx

@@ -33,6 +33,7 @@ import {
   continueStaticPrerender,
   continueDynamicHTMLResume,
   continueDynamicDataResume,
+  convertReadable,
 } from '../stream-utils'
 import { canSegmentBeOverridden } from '../../client/components/match-segments'
 import { stripInternalQueries } from '../internal-utils'
@@ -1020,21 +1021,22 @@ async function renderToHTMLOrFlightImpl(
       try {
         let { stream, postponed, resumed } = await renderer.render(children)
 
-        if (
-          process.env.NEXT_RUNTIME === 'nodejs' &&
-          !(stream instanceof ReadableStream)
-        ) {
-          const { Readable } = require('node:stream')
-          stream = Readable.toWeb(stream) as ReadableStream<Uint8Array>
-        }
+        // if (
+        //   process.env.NEXT_RUNTIME === 'nodejs' &&
+        //   !(stream instanceof ReadableStream)
+        // ) {
+        //   const { Readable } = require('node:stream')
+        //   stream = Readable.toWeb(stream) as ReadableStream<Uint8Array>
+        // }
 
-        // TODO (@Ethan-Arrowood): Remove this when stream utilities support both stream types.
-        if (!(stream instanceof ReadableStream)) {
-          throw new Error("Invariant: stream isn't a ReadableStream")
-        }
+        // // TODO (@Ethan-Arrowood): Remove this when stream utilities support both stream types.
+        // if (!(stream instanceof ReadableStream)) {
+        //   throw new Error("Invariant: stream isn't a ReadableStream")
+        // }
 
         const prerenderState = staticGenerationStore.prerenderState
         if (prerenderState) {
+          stream = convertReadable(stream)
           /**
            * When prerendering there are three outcomes to consider
            *
@@ -1175,6 +1177,7 @@ async function renderToHTMLOrFlightImpl(
             }
           }
         } else if (renderOpts.postponed) {
+          stream = convertReadable(stream)
           // This is a continuation of either an Incomplete or Dynamic Data Prerender.
           const inlinedDataStream = createInlinedDataReadableStream(
             dataStream,
@@ -1198,21 +1201,22 @@ async function renderToHTMLOrFlightImpl(
             }
           }
         } else {
+          const resultStream = await continueFizzStream(stream, {
+            inlinedDataStream: createInlinedDataReadableStream(
+              dataStream,
+              nonce,
+              formState
+            ),
+            isStaticGeneration: isStaticGeneration || generateStaticHTML,
+            getServerInsertedHTML,
+            serverInsertedHTMLToHead: true,
+            validateRootLayout,
+          })
           // This may be a static render or a dynamic render
           // @TODO factor this further to make the render types more clearly defined and remove
           // the deluge of optional params that passed to configure the various behaviors
           return {
-            stream: await continueFizzStream(stream, {
-              inlinedDataStream: createInlinedDataReadableStream(
-                dataStream,
-                nonce,
-                formState
-              ),
-              isStaticGeneration: isStaticGeneration || generateStaticHTML,
-              getServerInsertedHTML,
-              serverInsertedHTMLToHead: true,
-              validateRootLayout,
-            }),
+            stream: convertReadable(resultStream),
           }
         }
       } catch (err) {
@@ -1318,45 +1322,46 @@ async function renderToHTMLOrFlightImpl(
             },
           })
 
-          if (
-            process.env.NEXT_RUNTIME === 'nodejs' &&
-            !(fizzStream instanceof ReadableStream)
-          ) {
-            const { Readable } = require('node:stream')
-
-            fizzStream = Readable.toWeb(
-              fizzStream
-            ) as ReadableStream<Uint8Array>
-          }
-
-          // TODO (@Ethan-Arrowood): Remove this when stream utilities support both stream types.
-          if (!(fizzStream instanceof ReadableStream)) {
-            throw new Error("Invariant: stream isn't a ReadableStream")
-          }
-
+          // if (
+          //   process.env.NEXT_RUNTIME === 'nodejs' &&
+          //   !(fizzStream instanceof ReadableStream)
+          // ) {
+          //   const { Readable } = require('node:stream')
+
+          //   fizzStream = Readable.toWeb(
+          //     fizzStream
+          //   ) as ReadableStream<Uint8Array>
+          // }
+
+          // // TODO (@Ethan-Arrowood): Remove this when stream utilities support both stream types.
+          // if (!(fizzStream instanceof ReadableStream)) {
+          //   throw new Error("Invariant: stream isn't a ReadableStream")
+          // }
+
+          const resultStream = await continueFizzStream(fizzStream, {
+            inlinedDataStream: createInlinedDataReadableStream(
+              // This is intentionally using the readable datastream from the
+              // main render rather than the flight data from the error page
+              // render
+              dataStream,
+              nonce,
+              formState
+            ),
+            isStaticGeneration,
+            getServerInsertedHTML: makeGetServerInsertedHTML({
+              polyfills,
+              renderServerInsertedHTML,
+              serverCapturedErrors: [],
+              basePath: renderOpts.basePath,
+            }),
+            serverInsertedHTMLToHead: true,
+            validateRootLayout,
+          })
           return {
             // Returning the error that was thrown so it can be used to handle
             // the response in the caller.
             err,
-            stream: await continueFizzStream(fizzStream, {
-              inlinedDataStream: createInlinedDataReadableStream(
-                // This is intentionally using the readable datastream from the
-                // main render rather than the flight data from the error page
-                // render
-                dataStream,
-                nonce,
-                formState
-              ),
-              isStaticGeneration,
-              getServerInsertedHTML: makeGetServerInsertedHTML({
-                polyfills,
-                renderServerInsertedHTML,
-                serverCapturedErrors: [],
-                basePath: renderOpts.basePath,
-              }),
-              serverInsertedHTMLToHead: true,
-              validateRootLayout,
-            }),
+            stream: convertReadable(resultStream),
           }
         } catch (finalErr: any) {
           if (

packages/next/src/server/stream-utils/index.d.ts

@@ -16,5 +16,48 @@ export function renderToInitialFizzStream(
 export function renderToString(element: React.ReactElement): Promise<string>
 
 export function streamToString(
-  stream: Readable | ReadableStream
+  stream: Readable | ReadableStream<Uint8Array>
 ): Promise<string>
+
+export function chainStreams(
+  ...streams: ReadableStream<Uint8Array>[]
+): ReadableStream<Uint8Array>
+export function chainStreams(...streams: Readable[]): Readable
+
+export function convertReadable(
+  stream: Readable | ReadableStream<Uint8Array>
+): ReadableStream<Uint8Array>
+
+export function continueFizzStream(
+  stream: Readable,
+  options: {
+    inlinedDataStream?: Readable
+    isStaticGeneration: boolean
+    getServerInsertedHTML?: () => Promise<string>
+    serverInsertedHTMLToHead: boolean
+    validateRootLayout?: boolean
+    suffix?: string
+  }
+): Promise<Readable>
+export function continueFizzStream(
+  stream: ReadableStream<Uint8Array>,
+  options: {
+    inlinedDataStream?: ReadableStream<Uint8Array>
+    isStaticGeneration: boolean
+    getServerInsertedHTML?: () => Promise<string>
+    serverInsertedHTMLToHead: boolean
+    validateRootLayout?: boolean
+    suffix?: string
+  }
+): Promise<ReadableStream<Uint8Array>>
+export function continueFizzStream(
+  stream: Readable | ReadableStream<Uint8Array>,
+  options: {
+    inlinedDataStream?: Readable | ReadableStream<Uint8Array>
+    isStaticGeneration: boolean
+    getServerInsertedHTML?: () => Promise<string>
+    serverInsertedHTMLToHead: boolean
+    validateRootLayout?: boolean
+    suffix?: string
+  }
+): Promise<Readable | ReadableStream<Uint8Array>>

packages/next/src/server/stream-utils/stream-utils.edge.test.tsx

@@ -0,0 +1,45 @@
+import { createBufferedTransformStream } from './stream-utils.edge'
+import { renderToReadableStream } from 'react-dom/server.edge'
+import { Suspense } from 'react'
+
+function App() {
+  const Data = async () => {
+    const data = await Promise.resolve('1')
+    return <h2>{data}</h2>
+  }
+
+  return (
+    <html>
+      <head>
+        <title>My App</title>
+      </head>
+      <body>
+        <h1>Hello, World!</h1>
+        <Suspense fallback={<h2>Fallback</h2>}>
+          <Data />
+        </Suspense>
+      </body>
+    </html>
+  )
+}
+
+async function createInput(app = <App />): Promise<ReadableStream<Uint8Array>> {
+  const stream = await renderToReadableStream(app)
+  await stream.allReady
+  return stream
+}
+
+describe('createBufferedTransformStream', () => {
+  it('should return a TransformStream that buffers input chunks across rendering boundaries', async () => {
+    const input = await createInput()
+    const actualStream = input.pipeThrough(createBufferedTransformStream())
+
+    const actualChunks = []
+    // @ts-expect-error
+    for await (const chunks of actualStream) {
+      actualChunks.push(chunks)
+    }
+
+    expect(actualChunks.length).toBe(1)
+  })
+})

packages/next/src/server/stream-utils/stream-utils.edge.ts

@@ -657,3 +657,9 @@ export async function continueDynamicDataResume(
       .pipeThrough(createMoveSuffixStream(closeTag))
   )
 }
+
+export function convertReadable(
+  stream: ReadableStream<Uint8Array>
+): ReadableStream<Uint8Array> {
+  return stream
+}

packages/next/src/server/stream-utils/stream-utils.node.test.tsx

@@ -0,0 +1,85 @@
+import {
+  continueFizzStream,
+  createBufferedTransformStream,
+} from './stream-utils.node'
+import { PassThrough } from 'node:stream'
+import { renderToPipeableStream } from 'react-dom/server.node'
+import { Suspense } from 'react'
+
+function App() {
+  const Data = async () => {
+    const data = await Promise.resolve('1')
+    return <h2>{data}</h2>
+  }
+
+  return (
+    <html>
+      <head>
+        <title>My App</title>
+      </head>
+      <body>
+        <h1>Hello, World!</h1>
+        <Suspense fallback={<h2>Fallback</h2>}>
+          <Data />
+        </Suspense>
+      </body>
+    </html>
+  )
+}
+
+function createInput(app = <App />): Promise<PassThrough> {
+  return new Promise((resolve, reject) => {
+    const { pipe } = renderToPipeableStream(app, {
+      onAllReady() {
+        resolve(pipe(new PassThrough()))
+      },
+      onShellError(error) {
+        reject(error)
+      },
+    })
+  })
+}
+
+describe('createBufferedTransformStream', () => {
+  it('should return a TransformStream that buffers input chunks across rendering boundaries', (done) => {
+    createInput().then((input) => {
+      const stream = input.pipe(createBufferedTransformStream())
+      const actualChunks = []
+      stream.on('data', (chunk) => {
+        actualChunks.push(chunk)
+      })
+
+      stream.resume()
+
+      stream.on('end', () => {
+        expect(actualChunks.length).toBe(1)
+        done()
+      })
+    })
+  })
+})
+
+describe('continueFizzStream', () => {
+  it.only('should passthrough render stream and buffered transform stream', (done) => {
+    createInput().then((input) => {
+      continueFizzStream(input, {
+        isStaticGeneration: false,
+        serverInsertedHTMLToHead: false,
+      }).then((stream) => {
+        const actualChunks: Uint8Array[] = []
+        stream.on('data', (chunk) => {
+          actualChunks.push(chunk)
+        })
+
+        stream.resume()
+
+        stream.on('end', () => {
+          console.log('ended')
+          expect(actualChunks.length).toBe(2)
+          console.log(actualChunks[0].toString())
+          done()
+        })
+      })
+    })
+  })
+})