Merge pull request #8613 from QwikDev/v2-fix-node-crash

fix: node crash after not handled aborted connections

Author: Varixo

.changeset/dry-rabbits-hang.md

@@ -0,0 +1,5 @@
+---
+'@qwik.dev/router': patch
+---
+
+fix: handle aborted Node response streams without crashing and resolve the Node response stream contract

packages/qwik-router/src/middleware/node/http.ts

@@ -31,10 +31,23 @@ export function getUrl(req: IncomingMessage | Http2ServerRequest, origin: string
   return normalizeUrl((req as any).originalUrl || req.url || '/', origin);
 }
 
-// when the user refreshes or cancels the stream there will be an error
-function isIgnoredError(message = '') {
-  const ignoredErrors = ['The stream has been destroyed', 'write after end'];
-  return ignoredErrors.some((ignored) => message.includes(ignored));
+// Client disconnects are expected during streaming SSR: the socket is gone,
+// so the adapter should stop writing instead of surfacing them as app errors.
+function isClientAbortWriteError(error: unknown) {
+  if (!(error instanceof Error)) {
+    return false;
+  }
+  const code = (error as Error & { code?: string }).code;
+  if (
+    code === 'EPIPE' ||
+    code === 'ECONNRESET' ||
+    code === 'ERR_STREAM_DESTROYED' ||
+    code === 'ERR_STREAM_WRITE_AFTER_END'
+  ) {
+    return true;
+  }
+
+  return false;
 }
 
 // ensure no HTTP/2-specific headers are being set
@@ -46,20 +59,39 @@ export function normalizeUrl(url: string, base: string) {
 
 function createNodeResponseSink(res: ServerResponse) {
   let closed = res.closed || res.destroyed;
+  let responseError: unknown;
+  let resolveClosed: () => void;
   const closedPromise = closed
     ? Promise.resolve()
     : new Promise<void>((resolve) => {
+        resolveClosed = resolve;
         res.once('close', () => {
           closed = true;
           resolve();
         });
       });
 
+  res.on('error', (error) => {
+    if (responseError === undefined) {
+      responseError = error;
+    }
+    if (!closed) {
+      closed = true;
+      resolveClosed();
+    }
+  });
+
+  const shouldPropagateResponseError = () =>
+    responseError !== undefined && !isClientAbortWriteError(responseError);
+
   const write = (chunk: Uint8Array) => {
     if (closed || res.closed || res.destroyed) {
       // If the response has already been closed or destroyed (for example the client has disconnected)
       // then writing into it will cause an error. So just stop writing since no one
       // is listening.
+      if (shouldPropagateResponseError()) {
+        return Promise.reject(responseError);
+      }
       return;
     }
 
@@ -76,21 +108,33 @@ function createNodeResponseSink(res: ServerResponse) {
         setImmediate(resolve);
       };
 
+      const finishClosed = () => {
+        closed = true;
+        finish();
+      };
+
       const fail = (error: unknown) => {
         if (settled) {
           return;
         }
         settled = true;
+        closed = true;
         reject(error);
       };
 
-      closedPromise.then(finish);
+      closedPromise.then(() => {
+        if (shouldPropagateResponseError()) {
+          fail(responseError);
+          return;
+        }
+        finish();
+      });
 
       try {
         res.write(chunk, (error) => {
           if (error) {
-            if (isIgnoredError(error.message)) {
-              finish();
+            if (isClientAbortWriteError(error)) {
+              finishClosed();
               return;
             }
             fail(error);
@@ -99,8 +143,8 @@ function createNodeResponseSink(res: ServerResponse) {
           finish();
         });
       } catch (error) {
-        if (error instanceof Error && isIgnoredError(error.message)) {
-          finish();
+        if (isClientAbortWriteError(error)) {
+          finishClosed();
           return;
         }
         fail(error);
@@ -110,13 +154,52 @@ function createNodeResponseSink(res: ServerResponse) {
 
   const close = () => {
     if (closed || res.closed || res.destroyed) {
+      if (shouldPropagateResponseError()) {
+        return Promise.reject(responseError);
+      }
       return;
     }
 
-    return new Promise<void>((resolve) => {
-      res.end(() => {
+    return new Promise<void>((resolve, reject) => {
+      let settled = false;
+
+      const finish = () => {
+        if (settled) {
+          return;
+        }
+        settled = true;
+        closed = true;
         resolve();
+      };
+
+      const fail = (error: unknown) => {
+        if (settled) {
+          return;
+        }
+        settled = true;
+        closed = true;
+        reject(error);
+      };
+
+      closedPromise.then(() => {
+        if (shouldPropagateResponseError()) {
+          fail(responseError);
+          return;
+        }
+        finish();
       });
+
+      try {
+        res.end(() => {
+          finish();
+        });
+      } catch (error) {
+        if (isClientAbortWriteError(error)) {
+          finish();
+          return;
+        }
+        fail(error);
+      }
     });
   };
 
@@ -161,16 +244,20 @@ export async function fromNodeHttp(
 
   const body = req.method === 'HEAD' || req.method === 'GET' ? undefined : getRequestBody();
   const controller = new AbortController();
+  const abort = () => {
+    controller.abort();
+  };
   const options = {
     method: req.method,
     headers: requestHeaders,
     body: body as any,
     signal: controller.signal,
     duplex: 'half' as any,
   };
-  res.on('close', () => {
-    controller.abort();
-  });
+  req.on('aborted', abort);
+  req.on('error', abort);
+  res.on('close', abort);
+  res.on('error', abort);
   const serverRequestEv: ServerRequestEvent<boolean> = {
     mode,
     url,
@@ -180,7 +267,7 @@ export async function fromNodeHttp(
         return process.env[key];
       },
     },
-    getWritableStream: (status, headers, cookies) => {
+    getWritableStream: (status, headers, cookies, resolve) => {
       res.statusCode = status;
       const sink = createNodeResponseSink(res);
 
@@ -199,6 +286,8 @@ export async function fromNodeHttp(
         console.error(err);
       }
 
+      resolve(true);
+
       return new WritableStream<Uint8Array>({
         write(chunk) {
           return sink.write(chunk);