fix(@angular/ssr): settle writeResponseToNodeResponse when client disconnects

When a client disconnects while a response is backpressured or streaming,
the Node response (`ServerResponse` or `Http2ServerResponse`) is closed or
destroyed without emitting a `drain` event. Previously, this caused
`writeResponseToNodeResponse()` to park indefinitely waiting for `drain`
and never settle.

This change monitors whether the Node response is closed or destroyed
and removes event listeners, cancels the reader, and resolves the
returned Promise when a client disconnect occurs.

Closes #33719

TAG=agy
CONV=12c9fd92-1bdb-4e84-8c35-a43b5b69e75d

Author: alan-agius4

packages/angular/ssr/node/src/response.ts

@@ -9,6 +9,21 @@
 import type { ServerResponse } from 'node:http';
 import type { Http2ServerResponse } from 'node:http2';
 
+/**
+ * Checks whether a Node.js `ServerResponse` or `Http2ServerResponse` is destroyed or closed.
+ */
+function isResponseDestroyedOrClosed(destination: ServerResponse | Http2ServerResponse): boolean {
+  return (
+    Boolean(destination.destroyed) ||
+    Boolean(destination.closed) ||
+    Boolean(destination.writableEnded) ||
+    ('stream' in destination &&
+      (!destination.stream ||
+        Boolean(destination.stream.destroyed) ||
+        Boolean(destination.stream.closed)))
+  );
+}
+
 /**
  * Streams a web-standard `Response` into a Node.js `ServerResponse`
  * or `Http2ServerResponse`.
@@ -24,6 +39,10 @@ export async function writeResponseToNodeResponse(
   source: Response,
   destination: ServerResponse | Http2ServerResponse,
 ): Promise<void> {
+  if (isResponseDestroyedOrClosed(destination)) {
+    return;
+  }
+
   const { status, headers, body } = source;
   destination.statusCode = status;
 
@@ -48,40 +67,89 @@ export async function writeResponseToNodeResponse(
   }
 
   if (!body) {
-    destination.end();
+    if (!isResponseDestroyedOrClosed(destination)) {
+      destination.end();
+    }
 
     return;
   }
 
-  try {
-    const reader = body.getReader();
-
-    destination.on('close', () => {
-      reader.cancel().catch((error) => {
-        // eslint-disable-next-line no-console
-        console.error(
-          `An error occurred while writing the response body for: ${destination.req.url}.`,
-          error,
-        );
-      });
+  let isClosed = isResponseDestroyedOrClosed(destination);
+  const isDestroyedOrClosed = () => isClosed || isResponseDestroyedOrClosed(destination);
+
+  let readerCancelled = false;
+  const reader = body.getReader();
+  const cancelReader = (error?: unknown) => {
+    if (readerCancelled) {
+      return;
+    }
+    readerCancelled = true;
+    isClosed = true;
+    reader.cancel(error).catch((err) => {
+      // eslint-disable-next-line no-console
+      console.error(
+        `An error occurred while writing the response body for: ${destination.req.url}.`,
+        err,
+      );
     });
+  };
+
+  destination.once('close', cancelReader);
+  destination.once('error', cancelReader);
 
+  try {
     // eslint-disable-next-line no-constant-condition
     while (true) {
+      if (isDestroyedOrClosed()) {
+        cancelReader();
+        break;
+      }
+
       const { done, value } = await reader.read();
-      if (done) {
-        destination.end();
+      if (done || isDestroyedOrClosed()) {
+        if (!isDestroyedOrClosed()) {
+          destination.end();
+        } else {
+          cancelReader();
+        }
         break;
       }
 
       const canContinue = (destination as ServerResponse).write(value);
       if (canContinue === false) {
         // Explicitly check for `false`, as AWS may return `undefined` even though this is not valid.
         // See: https://github.com/CodeGenieApp/serverless-express/issues/683
-        await new Promise<void>((resolve) => destination.once('drain', resolve));
+        await new Promise<void>((resolve) => {
+          if (isDestroyedOrClosed()) {
+            resolve();
+            return;
+          }
+
+          const onDrain = () => {
+            destination.off('close', onClose);
+            destination.off('error', onClose);
+            resolve();
+          };
+
+          const onClose = () => {
+            destination.off('drain', onDrain);
+            destination.off('error', onClose);
+            cancelReader();
+            resolve();
+          };
+
+          destination.once('drain', onDrain);
+          destination.once('close', onClose);
+          destination.once('error', onClose);
+        });
       }
     }
   } catch {
-    destination.end('Internal server error.');
+    if (!isDestroyedOrClosed()) {
+      destination.end('Internal server error.');
+    }
+  } finally {
+    destination.off('close', cancelReader);
+    destination.off('error', cancelReader);
   }
 }

packages/angular/ssr/node/test/response_http1_spec.ts

@@ -107,4 +107,54 @@ describe('writeResponseToNodeResponse (HTTP/1.1)', () => {
 
     expect(response.headers['set-cookie']).toEqual(cookieValue);
   });
+
+  it('should resolve and cancel reader when client disconnects while response is backpressured', async () => {
+    let resolveWritePromise: () => void;
+    const writePromise = new Promise<void>((resolve) => {
+      resolveWritePromise = resolve;
+    });
+
+    let readerCancelled = false;
+    const largeChunk = 'x'.repeat(1024 * 1024 * 4); // 4MB to ensure backpressure
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(largeChunk);
+        controller.enqueue(largeChunk);
+      },
+      cancel() {
+        readerCancelled = true;
+      },
+    });
+
+    server.once('request', (req, nodeResponse) => {
+      writeResponseToNodeResponse(new Response(stream), nodeResponse).finally(() => {
+        resolveWritePromise();
+      });
+    });
+
+    await new Promise<void>((resolve) => {
+      const { port } = server.address() as AddressInfo;
+      const clientRequest = requestCb(
+        {
+          host: 'localhost',
+          port,
+        },
+        (response) => {
+          response.once('data', () => {
+            clientRequest.destroy();
+            resolve();
+          });
+        },
+      );
+
+      clientRequest.on('error', () => {
+        // Expected when destroying the socket
+      });
+
+      clientRequest.end();
+    });
+
+    await writePromise;
+    expect(readerCancelled).toBeTrue();
+  });
 });

packages/angular/ssr/node/test/response_http2_spec.ts

@@ -116,4 +116,54 @@ describe('writeResponseToNodeResponse (HTTP/2)', () => {
 
     expect(resHeaders['set-cookie']).toEqual(cookieValue);
   });
+
+  it('should resolve and cancel reader when client disconnects while response is backpressured', async () => {
+    let resolveWritePromise: () => void;
+    const writePromise = new Promise<void>((resolve) => {
+      resolveWritePromise = resolve;
+    });
+
+    let readerCancelled = false;
+    const largeChunk = 'x'.repeat(1024 * 1024); // 1MB to exceed HTTP/2 flow control window
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(largeChunk);
+        controller.enqueue(largeChunk);
+      },
+      cancel() {
+        readerCancelled = true;
+      },
+    });
+
+    server.once('request', (req, nodeResponse) => {
+      writeResponseToNodeResponse(new Response(stream), nodeResponse).finally(() => {
+        resolveWritePromise();
+      });
+    });
+
+    await new Promise<void>((resolve) => {
+      const { port } = server.address() as AddressInfo;
+      const client = connect(`http://localhost:${port}`);
+      const req = client.request({
+        ':path': '/',
+      });
+
+      req.once('response', () => {
+        req.once('data', () => {
+          req.destroy();
+          client.destroy();
+          resolve();
+        });
+      });
+
+      req.on('error', () => {
+        // Expected when destroying/closing the stream
+      });
+
+      req.end();
+    });
+
+    await writePromise;
+    expect(readerCancelled).toBeTrue();
+  });
 });