fix(adapter): do not let volatile packets break connection state recovery

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

@@ -466,7 +466,9 @@ export class SessionAwareAdapter extends Adapter {
     const missedPackets = [];
     for (let i = index + 1; i < this.packets.length; i++) {
       const packet = this.packets[i];
-      if (shouldIncludePacket(session.rooms, packet.opts)) {
+      // volatile packets are part of the offset chain but are never replayed
+      const isVolatile = packet.opts.flags?.volatile !== undefined;
+      if (!isVolatile && shouldIncludePacket(session.rooms, packet.opts)) {
         missedPackets.push(packet.data);
       }
     }
@@ -478,15 +480,16 @@ export class SessionAwareAdapter extends Adapter {
 
   override broadcast(packet: any, opts: BroadcastOptions) {
     const isEventPacket = packet.type === 2;
-    // packets with acknowledgement are not stored because the acknowledgement function cannot be serialized and
-    // restored on another server upon reconnection
+    // packets with acknowledgement are not stored because the acknowledgement function cannot be
+    // serialized and restored on another server upon reconnection
     const withoutAcknowledgement = packet.id === undefined;
-    const notVolatile = opts.flags?.volatile === undefined;
-    if (isEventPacket && withoutAcknowledgement && notVolatile) {
+    if (isEventPacket && withoutAcknowledgement) {
       const id = yeast();
       // the offset is stored at the end of the data array, so the client knows the ID of the last packet it has
       // processed (and the format is backward-compatible)
       packet.data.push(id);
+      // volatile packets are included in the offset chain so that the client always sends a known
+      // offset upon reconnection, but they are excluded from the replay in restoreSession()
       this.packets.push({
         id,
         opts,

packages/socket.io-adapter/test/index.ts

@@ -551,5 +551,67 @@ describe("socket.io-adapter", () => {
 
       expect(session).to.be(null);
     });
+
+    it("should still restore a session after the client received a volatile packet", async () => {
+      const adapter = new SessionAwareAdapter({
+        server: {
+          encoder: {
+            encode(packet) {
+              return packet;
+            },
+          },
+          opts: {
+            connectionStateRecovery: {
+              maxDisconnectionDuration: 5000,
+            },
+          },
+        },
+      });
+
+      adapter.persistSession({
+        sid: "abc",
+        pid: "def",
+        data: "ghi",
+        rooms: ["r1"],
+      });
+
+      const packetData = ["hello"];
+
+      adapter.broadcast(
+        {
+          nsp: "/",
+          type: 2,
+          data: packetData,
+        },
+        {
+          rooms: new Set(),
+          except: new Set(),
+        },
+      );
+
+      const volatileData = ["status", "connected"];
+
+      adapter.broadcast(
+        {
+          nsp: "/",
+          type: 2,
+          data: volatileData,
+        },
+        {
+          rooms: new Set(),
+          except: new Set(),
+          flags: {
+            volatile: true,
+          },
+        },
+      );
+
+      // the adapter appends its own offset at the end of the data array, so the client stores
+      // that offset instead of the last argument of the event itself
+      const session = await adapter.restoreSession("def", volatileData[2]);
+
+      expect(session).to.not.be(null);
+      expect(session.missedPackets.length).to.eql(0);
+    });
   });
 });