diff --git a/packages/engine.io/lib/transports/polling.ts b/packages/engine.io/lib/transports/polling.ts index f48f0cd6b..38b30a512 100644 --- a/packages/engine.io/lib/transports/polling.ts +++ b/packages/engine.io/lib/transports/polling.ts @@ -322,6 +322,15 @@ 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; @@ -329,11 +338,11 @@ export class Polling extends Transport { 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); diff --git a/packages/engine.io/test/compression-abort.js b/packages/engine.io/test/compression-abort.js new file mode 100644 index 000000000..e56534821 --- /dev/null +++ b/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 + }); + }, + ); + }); +});