From 7155a8a027ee74ae667b0a23ee8596e9078478c7 Mon Sep 17 00:00:00 2001 From: Mark Russell <39885340+mark1russell7@users.noreply.github.com> Date: Sat, 14 Nov 2026 13:08:06 -0500 Subject: [PATCH] fix(eio-client): abort the pending polling requests once the close packet is sent When the polling transport was closed, it sent a "close" packet, but it did not abort the requests which were still pending, like the long-polling request. If the server (or a proxy) never answered it, the request stayed open after the "close" event, holding a connection (and, in Node.js, preventing the process from exiting). The pending requests are now aborted once the close packet is sent (or cannot be sent). They are not aborted before, so that the server still sees a clean close. Note: with the "xmlhttprequest-ssl" package (Node.js), the requests are not aborted, because aborting a pending request throws an uncaught exception (see https://github.com/mjwwit/node-XMLHttpRequest/issues/37). Co-Authored-By: Claude Opus 5.5 --- .../lib/transports/polling-fetch.ts | 18 ++++++++- .../lib/transports/polling-xhr.node.ts | 5 +++ .../lib/transports/polling-xhr.ts | 29 +++++++++++--- .../lib/transports/polling.ts | 22 ++++++++++- packages/engine.io-client/test/node.js | 38 +++++++++++++++++++ 5 files changed, 103 insertions(+), 9 deletions(-) diff --git a/packages/engine.io-client/lib/transports/polling-fetch.ts b/packages/engine.io-client/lib/transports/polling-fetch.ts index 7d85b66c62..64fffa8274 100644 --- a/packages/engine.io-client/lib/transports/polling-fetch.ts +++ b/packages/engine.io-client/lib/transports/polling-fetch.ts @@ -10,6 +10,9 @@ import { Polling } from "./polling.js"; * @see https://nodejs.org/api/globals.html#fetch */ export class Fetch extends Polling { + private _abortController = + typeof AbortController === "function" ? new AbortController() : null; + override doPoll() { this._fetch() .then((res) => { @@ -24,20 +27,30 @@ export class Fetch extends Polling { }); } - override doWrite(data: string, callback: () => void) { + override doWrite( + data: string, + callback: () => void, + errorCallback?: () => void, + ) { this._fetch(data) .then((res) => { if (!res.ok) { - return this.onError("fetch write error", res.status, res); + this.onError("fetch write error", res.status, res); + return errorCallback && errorCallback(); } callback(); }) .catch((err) => { this.onError("fetch write error", err); + errorCallback && errorCallback(); }); } + protected override abortRequests() { + this._abortController?.abort(); + } + private _fetch(data?: string) { const isPost = data !== undefined; const headers = new Headers(this.opts.extraHeaders); @@ -53,6 +66,7 @@ export class Fetch extends Polling { body: isPost ? data : null, headers, credentials: this.opts.withCredentials ? "include" : "omit", + signal: this._abortController?.signal, }).then((res) => { // @ts-ignore getSetCookie() was added in Node.js v19.7.0 this.socket._cookieJar?.parseCookies(res.headers.getSetCookie()); diff --git a/packages/engine.io-client/lib/transports/polling-xhr.node.ts b/packages/engine.io-client/lib/transports/polling-xhr.node.ts index 172463fcf2..5730da5470 100644 --- a/packages/engine.io-client/lib/transports/polling-xhr.node.ts +++ b/packages/engine.io-client/lib/transports/polling-xhr.node.ts @@ -23,4 +23,9 @@ export class XHR extends BaseXHR { opts as RequestOptions, ); } + + protected override abortRequests() { + // note: the pending requests are not aborted, because aborting a pending request with the "xmlhttprequest-ssl" + // package throws an uncaught exception (see https://github.com/mjwwit/node-XMLHttpRequest/issues/37) + } } diff --git a/packages/engine.io-client/lib/transports/polling-xhr.ts b/packages/engine.io-client/lib/transports/polling-xhr.ts index 2196d2d220..b2bca0172b 100644 --- a/packages/engine.io-client/lib/transports/polling-xhr.ts +++ b/packages/engine.io-client/lib/transports/polling-xhr.ts @@ -16,6 +16,7 @@ export abstract class BaseXHR extends Polling { protected readonly xd: boolean; private pollXhr: any; + private _requests = new Set(); /** * XHR Polling constructor. @@ -56,14 +57,20 @@ export abstract class BaseXHR extends Polling { * @param {Function} fn - called upon flush. * @private */ - override doWrite(data: string, fn: () => void) { + override doWrite(data: string, fn: () => void, errorFn?: () => void) { const req = this.request({ method: "POST", data: data, }); - req.on("success", fn); + this._requests.add(req); + req.on("success", () => { + this._requests.delete(req); + fn(); + }); req.on("error", (xhrStatus, context) => { + this._requests.delete(req); this.onError("xhr post error", xhrStatus, context); + errorFn && errorFn(); }); } @@ -75,12 +82,22 @@ export abstract class BaseXHR extends Polling { override doPoll() { debug("xhr poll"); const req = this.request(); - req.on("data", this.onData.bind(this)); + this._requests.add(req); + req.on("data", (data) => { + this._requests.delete(req); + this.onData(data); + }); req.on("error", (xhrStatus, context) => { + this._requests.delete(req); this.onError("xhr poll error", xhrStatus, context); }); this.pollXhr = req; } + + protected override abortRequests() { + this._requests.forEach((req) => req.abort(true)); + this._requests.clear(); + } } interface RequestReservedEvents { @@ -282,10 +299,12 @@ export class Request extends Emitter< /** * Aborts the request. * + * @param {Boolean} force - whether the underlying XMLHttpRequest object should be aborted too (which closes the + * connection) * @package */ - public abort() { - this._cleanup(); + public abort(force?: boolean) { + this._cleanup(force); } } diff --git a/packages/engine.io-client/lib/transports/polling.ts b/packages/engine.io-client/lib/transports/polling.ts index 2f09e8021d..70d2b38e20 100644 --- a/packages/engine.io-client/lib/transports/polling.ts +++ b/packages/engine.io-client/lib/transports/polling.ts @@ -122,7 +122,14 @@ export abstract class Polling extends Transport { override doClose() { const close = () => { debug("writing close packet"); - this.write([{ type: "close" }]); + this.writable = false; + + encodePayload([{ type: "close" }], (data) => { + // once the close packet is sent (or cannot be sent), the pending requests (for example the long-polling + // request, if the server did not answer it) are not needed anymore + const onSettled = () => this.abortRequests(); + this.doWrite(data, onSettled, onSettled); + }); }; if ("open" === this.readyState) { @@ -174,6 +181,17 @@ export abstract class Polling extends Transport { return this.createUri(schema, query); } + /** + * Aborts the pending requests, once the transport is closed. + * + * @protected + */ + protected abortRequests() {} + abstract doPoll(); - abstract doWrite(data: string, callback: () => void); + abstract doWrite( + data: string, + callback: () => void, + errorCallback?: () => void, + ); } diff --git a/packages/engine.io-client/test/node.js b/packages/engine.io-client/test/node.js index 1f756a13bd..83276190d5 100644 --- a/packages/engine.io-client/test/node.js +++ b/packages/engine.io-client/test/node.js @@ -5,8 +5,46 @@ const { repeat } = require("./util"); const expect = require("expect.js"); const { parse } = require("../build/cjs/globals.node.js"); const { CookieJar } = require("../build/cjs/globals.node.js"); +const { createServer } = require("node:http"); +const { useFetch } = require("./support/env"); describe("node.js", () => { + it("should abort the long-polling request once the close packet is sent", function (done) { + if (!useFetch) { + // aborting a pending request with the "xmlhttprequest-ssl" package throws an uncaught exception (see + // https://github.com/mjwwit/node-XMLHttpRequest/issues/37) + return this.skip(); + } + + let socket; + + const httpServer = createServer((req, res) => { + if (req.method === "POST") { + req.resume(); + return req.on("end", () => res.end("ok")); + } + if (!new URL(req.url, "http://localhost").searchParams.has("sid")) { + return res.end( + '0{"sid":"abc","upgrades":[],"pingInterval":10000,"pingTimeout":5000,"maxPayload":1000000}', + ); + } + // the long-polling request is never answered + res.on("close", () => { + httpServer.close(); + done(); + }); + + // the socket is closed while the long-polling request is pending + socket.close(); + }); + + httpServer.listen(0, () => { + socket = new Socket(`http://localhost:${httpServer.address().port}`, { + transports: ["polling"], + }); + }); + }); + describe("autoRef option", () => { const runFixture = (filename, done) => execFile(