clean up

Author: darrachequesne

packages/socket.io-adapter/lib/cluster-adapter.ts

@@ -482,7 +482,7 @@ export abstract class ClusterAdapter extends Adapter {
     opts: BroadcastOptions,
     clientCountCallback: (clientCount: number) => void,
     ack: (...args: any[]) => void,
-  ): { cleanup: () => void } {
+  ) {
     const onlyLocal = opts?.flags?.local;
     if (!onlyLocal) {
       const requestId = randomId();

packages/socket.io-adapter/lib/in-memory-adapter.ts

@@ -199,7 +199,7 @@ export class Adapter extends EventEmitter {
     opts: BroadcastOptions,
     clientCountCallback: (clientCount: number) => void,
     ack: (...args: any[]) => void,
-  ): { cleanup: () => void } {
+  ) {
     const flags = opts.flags || {};
     const packetOpts = {
       preEncoded: true,
@@ -214,15 +214,12 @@ export class Adapter extends EventEmitter {
     const encodedPackets = this._encode(packet, packetOpts);
 
     let clientCount = 0;
-    const sentToSockets: any[] = [];
 
     this.apply(opts, (socket) => {
       // track the total number of acknowledgements that are expected
       clientCount++;
       // call the ack callback for each client response
       socket.acks.set(packet.id, ack);
-      // track sockets for cleanup on timeout
-      sentToSockets.push(socket);
 
       if (typeof socket.notifyOutgoingListeners === "function") {
         socket.notifyOutgoingListeners(packet);
@@ -233,17 +230,11 @@ export class Adapter extends EventEmitter {
 
     clientCountCallback(clientCount);
 
-    const packetId = packet.id;
     return {
       cleanup: () => {
-        // remove pending acks from all sockets on timeout
-        for (const socket of sentToSockets) {
-          if (socket.acks) {
-            socket.acks.delete(packetId);
-          }
-        }
-        // clear array to release socket references
-        sentToSockets.length = 0;
+        this.apply(opts, (socket) => {
+          socket.acks.delete(packet.id);
+        });
       },
     };
   }

packages/socket.io/lib/broadcast-operator.ts

@@ -232,14 +232,11 @@ export class BroadcastOperator<EmitEvents extends EventsMap, SocketData>
     const ack = data.pop() as (...args: any[]) => void;
     let timedOut = false;
     let responses: any[] = [];
-    let cleanup: (() => void) | undefined;
+    let cleanupPendingAcks: (() => void) | undefined;
 
     const timer = setTimeout(() => {
       timedOut = true;
-      // cleanup pending acks to prevent memory leak
-      if (cleanup) {
-        cleanup();
-      }
+      cleanupPendingAcks?.();
       ack.apply(this, [
         new Error("operation has timed out"),
         this.flags.expectSingleResponse ? null : responses,
@@ -284,9 +281,8 @@ export class BroadcastOperator<EmitEvents extends EventsMap, SocketData>
       },
     );
 
-    // store cleanup function for timeout handler
     if (result && typeof result.cleanup === "function") {
-      cleanup = result.cleanup;
+      cleanupPendingAcks = result.cleanup;
     }
 
     this.adapter.serverCount().then((serverCount) => {

packages/socket.io/test/messaging-many.ts

@@ -534,6 +534,11 @@ describe("messaging many", () => {
         // @ts-ignore
         expect(err.responses).to.contain(1, 2);
 
+        for (const [, serverSocket] of io.of("/").sockets) {
+          // @ts-ignore accessing private acks map to verify cleanup
+          expect(serverSocket.acks.size).to.be(0);
+        }
+
         success(done, io, socket1, socket2, socket3);
       }
     });

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

@@ -84,75 +84,4 @@ describe("timeout", () => {
       success(done, io, client);
     });
   });
-
-  it("should cleanup pending acks on broadcast timeout (memory leak fix)", (done) => {
-    const io = new Server(0);
-    const client = createClient(io, "/");
-
-    // Client does not acknowledge the event (simulates timeout scenario)
-    client.on("test-event", () => {
-      // intentionally not calling the callback
-    });
-
-    io.on("connection", async (socket) => {
-      socket.join("test-room");
-
-      // Get initial acks count (cast to any to access private property in test)
-      const initialAcksSize = (socket as any).acks.size;
-
-      try {
-        await io.timeout(50).to("test-room").emitWithAck("test-event", "data");
-        expect().fail("should have timed out");
-      } catch (err) {
-        expect(err).to.be.an(Error);
-
-        // After timeout, acks should be cleaned up (no memory leak)
-        // Wait a bit for cleanup to complete
-        setTimeout(() => {
-          expect((socket as any).acks.size).to.be(initialAcksSize);
-          success(done, io, client);
-        }, 10);
-      }
-    });
-  });
-
-  it("should cleanup pending acks on broadcast timeout with multiple clients", (done) => {
-    const io = new Server(0);
-    const client1 = createClient(io, "/");
-    const client2 = createClient(io, "/");
-
-    let connectedSockets: any[] = [];
-
-    // Clients do not acknowledge
-    client1.on("test-event", () => {});
-    client2.on("test-event", () => {});
-
-    io.on("connection", (socket) => {
-      socket.join("test-room");
-      connectedSockets.push(socket);
-
-      if (connectedSockets.length === 2) {
-        runTest();
-      }
-    });
-
-    async function runTest() {
-      const initialAcksSizes = connectedSockets.map((s) => s.acks.size);
-
-      try {
-        await io.timeout(50).to("test-room").emitWithAck("test-event", "data");
-        expect().fail("should have timed out");
-      } catch (err) {
-        expect(err).to.be.an(Error);
-
-        setTimeout(() => {
-          // All sockets should have their acks cleaned up
-          connectedSockets.forEach((socket, i) => {
-            expect(socket.acks.size).to.be(initialAcksSizes[i]);
-          });
-          success(done, io, client1, client2);
-        }, 10);
-      }
-    }
-  });
 });