fix(client): discard timed-out packets from the Engine.IO buffer

packages/engine.io-client/lib/socket.ts

@@ -689,6 +689,34 @@ export class SocketWithoutUpgrade extends Emitter<
     }
   }
 
+  /**
+   * Removes packets that have not been handed to the transport yet.
+   *
+   * Packets from one Socket.IO event share the same `options` object. If any of
+   * those packets is already part of the current flush batch, all related
+   * packets are left intact so a multi-packet payload cannot be partially sent.
+   *
+   * @param options - the options object shared by the packets to remove
+   * @private
+   */
+  /* private */ _removeFromWriteBuffer(options: WriteOptions) {
+    if (options == null) {
+      return;
+    }
+
+    for (let i = 0; i < this._prevBufferLen; i++) {
+      if (this.writeBuffer[i].options === options) {
+        return;
+      }
+    }
+
+    for (let i = this.writeBuffer.length - 1; i >= this._prevBufferLen; i--) {
+      if (this.writeBuffer[i].options === options) {
+        this.writeBuffer.splice(i, 1);
+      }
+    }
+  }
+
   /**
    * Flush write buffers.
    *

packages/engine.io-client/test/socket.js

@@ -13,6 +13,56 @@ const { repeat } = require("./util");
 describe("Socket", function () {
   this.timeout(10000);
 
+  describe("_removeFromWriteBuffer", () => {
+    it("should remove all matching packets that have not been flushed", () => {
+      const socket = Object.create(Socket.prototype);
+      const options = {};
+      const otherPacket = { options: {} };
+      socket.writeBuffer = [{ options }, otherPacket, { options }];
+      socket._prevBufferLen = 0;
+
+      socket._removeFromWriteBuffer(options);
+
+      expect(socket.writeBuffer).to.eql([otherPacket]);
+    });
+
+    it("should preserve all matching packets when one has been flushed", () => {
+      const socket = Object.create(Socket.prototype);
+      const options = {};
+      const packets = [{ options }, { options }];
+      socket.writeBuffer = packets.slice();
+      socket._prevBufferLen = 1;
+
+      socket._removeFromWriteBuffer(options);
+
+      expect(socket.writeBuffer).to.eql(packets);
+    });
+
+    it("should not remove already flushed packets or change the flush cursor", () => {
+      const socket = Object.create(Socket.prototype);
+      const options = {};
+      const flushed = { options: {} };
+      socket.writeBuffer = [flushed, { options }, { options }];
+      socket._prevBufferLen = 1;
+
+      socket._removeFromWriteBuffer(options);
+
+      expect(socket.writeBuffer).to.eql([flushed]);
+      expect(socket._prevBufferLen).to.be(1);
+    });
+
+    it("should ignore a missing options object", () => {
+      const socket = Object.create(Socket.prototype);
+      const packet = { options: {} };
+      socket.writeBuffer = [packet];
+      socket._prevBufferLen = 0;
+
+      socket._removeFromWriteBuffer(undefined);
+
+      expect(socket.writeBuffer).to.eql([packet]);
+    });
+  });
+
   describe("filterUpgrades", () => {
     it("should return only available transports", () => {
       const socket = new Socket({ transports: ["polling"] });

packages/socket.io-client/lib/socket.ts

@@ -440,7 +440,7 @@ export class Socket<
       debug("emitting packet with ack id %d", id);
 
       const ack = args.pop() as (...args: any[]) => void;
-      this._registerAckCallback(id, ack);
+      this._registerAckCallback(id, ack, packet);
       packet.id = id;
     }
 
@@ -465,7 +465,11 @@ export class Socket<
   /**
    * @private
    */
-  private _registerAckCallback(id: number, ack: (...args: any[]) => void) {
+  private _registerAckCallback(
+    id: number,
+    ack: (...args: any[]) => void,
+    packet: Packet & { options?: { compress?: boolean } },
+  ) {
     const timeout = this.flags.timeout ?? this._opts.ackTimeout;
     if (timeout === undefined) {
       this.acks[id] = ack;
@@ -475,6 +479,9 @@ export class Socket<
     // @ts-ignore
     const timer = this.io.setTimeoutFn(() => {
       delete this.acks[id];
+      // The packet may already have been written to the Engine.IO buffer
+      // (connection still "open" but transport temporarily unwritable).
+      this.io.engine?._removeFromWriteBuffer(packet.options);
       for (let i = 0; i < this.sendBuffer.length; i++) {
         if (this.sendBuffer[i].id === id) {
           debug("removing packet with ack id %d from the buffer", id);

packages/socket.io-client/test/socket.ts

@@ -596,6 +596,95 @@ describe("socket", () => {
       });
     });
 
+    it("should not flush a timed out event when the transport becomes writable again", () => {
+      return wrap((done) => {
+        const socket = io(BASE_URL + "/", {
+          forceNew: true,
+          transports: ["websocket"],
+        });
+
+        socket.on("connect", () => {
+          const engine = socket.io.engine;
+          engine.transport.writable = false;
+
+          const results: { one?: unknown; three?: unknown } = {};
+          let timedOut = false;
+
+          socket.emit("tracked", "one", (value) => {
+            results.one = value;
+            maybeDone();
+          });
+
+          socket.timeout(50).emit("tracked", "two", (err) => {
+            expect(err).to.be.an(Error);
+            timedOut = true;
+
+            engine.transport.writable = true;
+            // @ts-ignore flush remaining Engine.IO packets
+            engine.flush();
+          });
+
+          socket.timeout(1000).emit("tracked", "three", (err, value) => {
+            results.three = err || value;
+            maybeDone();
+          });
+
+          function maybeDone() {
+            if (results.one === undefined || results.three === undefined) {
+              return;
+            }
+
+            expect(timedOut).to.be(true);
+            expect(results.one).to.be("one");
+            expect(results.three).to.be("three");
+
+            socket.emit("getTracked", (events) => {
+              expect(events).to.eql(["one", "three"]);
+              success(done, socket);
+            });
+          }
+        });
+      });
+    });
+
+    it("should discard a timed out packet buffered by Engine.IO", () => {
+      return wrap((done) => {
+        const socket = io(BASE_URL + "/", {
+          forceNew: true,
+          transports: ["websocket"],
+        });
+
+        socket.on("connect", () => {
+          const engine = socket.io.engine;
+          engine.transport.writable = false;
+
+          const enginePackets = [];
+          const onPacketCreate = (packet) => {
+            enginePackets.push(packet);
+          };
+          engine.on("packetCreate", onPacketCreate);
+
+          socket.timeout(50).emit("tracked", Uint8Array.from([1]), (err) => {
+            expect(err).to.be.an(Error);
+            expect(enginePackets.length).to.be(2);
+            enginePackets.forEach((packet) => {
+              expect(engine.writeBuffer.indexOf(packet)).to.be(-1);
+            });
+
+            engine.transport.writable = true;
+            // @ts-ignore flush remaining Engine.IO packets
+            engine.flush();
+
+            socket.emit("getTracked", (events) => {
+              expect(events).to.eql([]);
+              success(done, socket);
+            });
+          });
+          engine.off("packetCreate", onPacketCreate);
+        });
+      });
+    });
+
     it("should timeout when the server does not acknowledge the event", () => {
       return wrap((done) => {
         const socket = io(BASE_URL + "/");

packages/socket.io-client/test/support/server.ts

@@ -60,6 +60,18 @@ export function createServer() {
       cb(arg);
     });
 
+    socket.on("tracked", (arg, cb) => {
+      socket.data.tracked = socket.data.tracked || [];
+      socket.data.tracked.push(arg);
+      if (typeof cb === "function") {
+        cb(arg);
+      }
+    });
+
+    socket.on("getTracked", (cb) => {
+      cb(socket.data.tracked || []);
+    });
+
     // ack tests
     socket.on("ack", () => {
       socket.emit("ack", (a, b) => {