fix(eio): run the polling write callback when the client aborts a compressed response (#5540)

Since [1] the write callback of doWrite() only ran on the response's
'finish' event. When the client aborted mid-stream, 'finish' never fired
and the zlib pipeline reported no error, so the callback was lost: the
pending request was not cleaned up and the send callbacks queued for that
batch were never invoked.

The callback now also runs on 'close', guarded so it still fires exactly
once.

Co-authored-by: chuanghiduoc <chuanghiduoc@users.noreply.github.com>

[1]: 91dd763

Author: chuanghiduoc

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
+        });
+      },
+    );
+  });
+});