Revert "Fix stream cancellation in RenderResult.pipe() and `sendRes…

…ponse()`" (#52277)

The test here seems to be failing after merge 

x-ref:
https://github.com/vercel/next.js/actions/runs/5467319221/jobs/9953529745
x-ref:
https://github.com/vercel/next.js/actions/runs/5467152519/jobs/9953132439

Reverts #52157

packages/next-swc/crates/next-core/js/src/entry/app/edge-page-bootstrap.ts

@@ -80,20 +80,11 @@ async function render(request: NextRequest, event: NextFetchEvent) {
   response.headers.append('Vary', RSC_VARY_HEADER)
 
   const writer = tranform.writable.getWriter()
-  const target = {
+  result.pipe({
     write: (chunk: Uint8Array) => writer.write(chunk),
     end: () => writer.close(),
     destroy: (reason?: Error) => writer.abort(reason),
-    closed: false,
-  }
-  const onClose = () => {
-    target.closed = true
-  }
-  // No, this cannot be replaced with `finally`, because early cancelling
-  // the stream will create a rejected promise, and finally will create an
-  // unhandled rejection.
-  writer.closed.then(onClose, onClose)
-  result.pipe(target)
+  })
 
   return response
 }

packages/next/src/server/render-result.ts

@@ -16,7 +16,6 @@ export interface PipeTarget {
   end: () => unknown
   flush?: () => unknown
   destroy: (err?: Error) => unknown
-  get closed(): boolean
 }
 
 export default class RenderResult {
@@ -112,25 +111,31 @@ export default class RenderResult {
         : () => {}
     const reader = this.response.getReader()
 
+    let shouldFatalError = false
     try {
-      while (true) {
-        const result = await reader.read()
-
-        if (res.closed || result.done) {
-          break
-        }
+      let result = await reader.read()
+      if (!result.done) {
+        // As we're going to write to the response, we should destroy the
+        // response if an error occurs.
+        shouldFatalError = true
+      }
 
+      while (!result.done) {
         // Write the data to the response.
         res.write(result.value)
 
         // Flush it to the client (if it supports flushing).
         flush()
+
+        // Read the next chunk.
+        result = await reader.read()
       }
 
       // We're done writing to the response, so we can end it.
       res.end()
     } catch (err) {
-      if (!res.closed) {
+      // If we've written to the response, we should destroy it.
+      if (shouldFatalError) {
         res.destroy(err as any)
       }
 

packages/next/src/server/send-response.ts

@@ -49,7 +49,6 @@ export async function sendResponse(
       const iterator = consumeUint8ArrayReadableStream(response.body)
       try {
         for await (const chunk of iterator) {
-          if (originalResponse.closed) break
           originalResponse.write(chunk)
         }
       } finally {

packages/next/src/server/stream-utils/node-web-streams-helper.ts

@@ -25,9 +25,8 @@ export const streamToBufferedResult = async (
     },
     end() {},
     destroy() {},
-    closed: false,
   }
-  await renderResult.pipe(writable)
+  await renderResult.pipe(writable as any)
   return renderChunks.join('')
 }
 

packages/next/src/server/web-server.ts

@@ -421,20 +421,14 @@ export default class NextWebServer extends BaseServer<WebServerOptions> {
 
     if (options.result.isDynamic) {
       const writer = res.transformStream.writable.getWriter()
-      const target = {
+      options.result.pipe({
         write: (chunk: Uint8Array) => writer.write(chunk),
         end: () => writer.close(),
         destroy: (err: Error) => writer.abort(err),
-        closed: false,
-      } as any
-      const onClose = () => {
-        target.closed = true
-      }
-      // No, this cannot be replaced with `finally`, because early cancelling
-      // the stream will create a rejected promise, and finally will create an
-      // unhandled rejection.
-      writer.closed.then(onClose, onClose)
-      options.result.pipe(target)
+        cork: () => {},
+        uncork: () => {},
+        // Not implemented: on/removeListener
+      } as any)
     } else {
       const payload = await options.result.toUnchunkedString()
       res.setHeader('Content-Length', String(byteLength(payload)))