fix(engine.io): catch synchronous middleware errors

packages/engine.io/lib/server.ts

@@ -397,17 +397,45 @@ export abstract class BaseServer extends EventEmitter {
 
     const apply = (i) => {
       debug("applying middleware n°%d", i + 1);
-      this.middlewares[i](req, res, (err?: any) => {
-        if (err) {
-          return callback(err);
+
+      // next(), a throw, or a rejection may each reach callback. Only the first
+      // one wins: a middleware that calls next() and then throws must not fire
+      // callback a second time. Errors thrown by next()'s own continuation are
+      // not middleware failures and are rethrown.
+      let settled = false;
+      let continuationError: any;
+
+      const done = (err?: any) => {
+        if (settled) {
+          debug("middleware already settled, ignoring extra callback");
+          return;
+        }
+        settled = true;
+
+        try {
+          if (err) {
+            callback(err);
+          } else if (i + 1 < this.middlewares.length) {
+            apply(i + 1);
+          } else {
+            callback();
+          }
+        } catch (e) {
+          continuationError = e;
+          throw e;
         }
+      };
 
-        if (i + 1 < this.middlewares.length) {
-          apply(i + 1);
-        } else {
-          callback();
+      try {
+        Promise.resolve(this.middlewares[i](req, res, done)).catch((err) => {
+          done(err);
+        });
+      } catch (e) {
+        if (continuationError) {
+          throw continuationError;
         }
-      });
+        done(e);
+      }
     };
 
     apply(0);
@@ -1042,6 +1070,16 @@ function abortRequest(
       ? errorContext.message
       : Server.errorMessages[errorCode];
 
+  // A middleware may have already committed the response before failing.
+  // Writing again throws "Cannot write headers after they are sent".
+  if (res.headersSent) {
+    debug("response headers already sent, skipping error response");
+    if (!res.writableEnded && !res.finished) {
+      res.end();
+    }
+    return;
+  }
+
   res.writeHead(statusCode, { "Content-Type": "application/json" });
   res.end(
     JSON.stringify({
@@ -1116,7 +1154,7 @@ const validHdrChars = [
   1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
   1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
   1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1  // ... 255
-]
+];
 
 function checkInvalidHeaderChar(val?: string) {
   val += "";

packages/engine.io/lib/userver.ts

@@ -1,4 +1,5 @@
 import debugModule from "debug";
+import { STATUS_CODES } from "http";
 import { AttachOptions, BaseServer, Server } from "./server";
 import { HttpRequest, HttpResponse, TemplatedApp } from "uWebSockets.js";
 import transports from "./transports-uws";
@@ -269,6 +270,15 @@ export class uServer extends BaseServer {
     errorCode: number,
     errorContext?: { message?: string },
   ) {
+    // uWS throws if HttpResponse is used after a successful end().
+    if (res instanceof ResponseWrapper && res.headersSent) {
+      debug("response headers already sent, skipping error response");
+      if (!res.writableEnded) {
+        res.end();
+      }
+      return;
+    }
+
     const statusCode =
       errorCode === Server.errors.FORBIDDEN
         ? "403 Forbidden"
@@ -291,21 +301,45 @@ export class uServer extends BaseServer {
 
 class ResponseWrapper {
   private statusWritten: boolean = false;
+  private ended = false;
   private headers = [];
   private isAborted = false;
 
   constructor(readonly res: HttpResponse) {}
 
+  public get headersSent() {
+    return this.statusWritten || this.ended;
+  }
+
+  public get writableEnded() {
+    return this.ended;
+  }
+
   public set statusCode(status: number) {
     if (!status) {
       return;
     }
-    // FIXME: handle all status codes?
-    this.writeStatus(status === 200 ? "200 OK" : "204 No Content");
+    const reason = STATUS_CODES[status] || "Unknown";
+    this.writeStatus(`${status} ${reason}`);
   }
 
-  public writeHead(status: number) {
-    this.statusCode = status;
+  public writeHead(status?: number, headers?: any, maybeHeaders?: any) {
+    if (typeof status === "number") {
+      this.statusCode = status;
+    }
+
+    const headerFields =
+      headers !== null && typeof headers === "object" ? headers : maybeHeaders;
+    if (
+      !headerFields ||
+      typeof headerFields !== "object" ||
+      Array.isArray(headerFields)
+    ) {
+      return;
+    }
+    Object.keys(headerFields).forEach((key) => {
+      this.setHeader(key, headerFields[key]);
+    });
   }
 
   public setHeader(key, value) {
@@ -354,15 +388,20 @@ class ResponseWrapper {
     });
   }
 
-  public end(data) {
-    if (this.isAborted) return;
+  public end(data?) {
+    if (this.isAborted || this.ended) return;
 
+    this.ended = true;
     this.res.cork(() => {
       if (!this.statusWritten) {
         // status will be inferred as "200 OK"
         this.writeBufferedHeaders();
       }
-      this.res.end(data);
+      if (data === undefined) {
+        this.res.end();
+      } else {
+        this.res.end(data);
+      }
     });
   }