fix(engine.io): run the polling write callback when the client aborts a compressed response

packages/engine.io/lib/transports/polling.ts

@@ -322,18 +322,27 @@ export class Polling extends Transport {
     const stream = compressionMethods[encoding](this.httpCompression);
 
     let isErrored = false;
+    let isDone = false;
+
+    const done = () => {
+      if (isDone || isErrored) {
+        return;
+      }
+      isDone = true;
+      callback();
+    };
 
     stream.on("error", (err) => {
       isErrored = true;
       this.res.end();
       callback(err);
     });
 
-    this.res.once("finish", () => {
-      if (!isErrored) {
-        callback();
-      }
-    });
+    // 'close' also fires after a normal completion, hence the guard: whatever
+    // happens, the write callback must run exactly once so the transport
+    // cleans up its request state and emits 'drain'.
+    this.res.once("finish", done);
+    this.res.once("close", done);
 
     stream.pipe(this.res);
     stream.end(data);

packages/engine.io/test/compression-abort.js

@@ -0,0 +1,76 @@
+/* eslint-disable standard/no-callback-literal */
+
+const http = require("http");
+const crypto = require("crypto");
+const cookieMod = require("cookie");
+const { listen } = require("./common");
+const expect = require("expect.js");
+
+function getSidFromResponse(res) {
+  const c = cookieMod.parse(res.headers["set-cookie"][0]);
+  return c[Object.keys(c)[0]];
+}
+
+describe("polling compression", () => {
+  let engine;
+
+  afterEach(() => {
+    if (engine && engine.httpServer) {
+      engine.httpServer.close();
+    }
+  });
+
+  it("should not lose the write callback when the client aborts a compressed response mid-stream", (done) => {
+    engine = listen(
+      {
+        cookie: true,
+        transports: ["polling"],
+        httpCompression: { threshold: 0 },
+        pingInterval: 60000,
+        pingTimeout: 60000,
+      },
+      (port) => {
+        // incompressible content so real bytes keep flowing to the socket
+        const chunk = crypto.randomBytes(1024 * 1024).toString("base64");
+        let sendCallbackCalled = false;
+
+        engine.on("connection", (c) => {
+          const spam = setInterval(() => {
+            if (c.readyState !== "open") {
+              clearInterval(spam);
+              return;
+            }
+            c.send(chunk, () => {
+              sendCallbackCalled = true;
+            });
+          }, 5);
+          setTimeout(() => clearInterval(spam), 2000);
+        });
+
+        http.get({ port, path: "/engine.io/?transport=polling" }, (res) => {
+          const sid = getSidFromResponse(res);
+          const pollReq = http.get(
+            {
+              port,
+              path: "/engine.io/?transport=polling&sid=" + sid,
+              headers: { "Accept-Encoding": "gzip, deflate" },
+            },
+            (pollRes) => {
+              // abort on headers, before any body byte is written
+              pollReq.destroy();
+              setTimeout(() => {
+                try {
+                  expect(sendCallbackCalled).to.be(true);
+                  done();
+                } catch (e) {
+                  done(e);
+                }
+              }, 1500);
+            },
+          );
+          pollReq.on("error", () => {}); // expected: socket hang up
+        });
+      },
+    );
+  });
+});