Make room departures cross-process safe

Claim persisted room departures with a MongoDB compare-and-set on the active membership revision. This prevents competing Socket.IO processes from duplicating leave side effects or overwriting a rejoin, while retaining the reconnect grace behavior.\n\nRefs #153

Author: coder13

docs/room-ownership.md

@@ -34,6 +34,12 @@ itself from the per-user socket group before the remaining-tab check, so two
 simultaneous explicit leaves cannot strand membership. If the owner later
 rejoins, they reclaim admin before the join is acknowledged.
 
+The final persisted departure is a MongoDB compare-and-set keyed by the active
+membership revision. This applies across Socket.IO processes: exactly one
+process can claim a leave, emit its room events, and record its metrics. A
+concurrent rejoin advances the revision, so a losing leave retries from current
+state and cannot overwrite the new membership.
+
 An empty room persists `admin: null`, but the established `UPDATE_ADMIN`
 Socket.IO event is emitted only for a non-null active admin. Spectators and
 other clients therefore never receive a null admin payload.

server/models/room.js

@@ -7,6 +7,22 @@ const { mirrorRoomChanges } = require('../postgres/dualWrite');
 const PASSWORD_SALT_ROUNDS = 10;
 const STALE_ROOM_LIFETIME_MS = 10 * 60 * 1000;
 
+const sameUser = (left, right) => left && right
+  && String(left.id) === String(right.id);
+
+const selectRoomAdmin = ({ usersInRoom, owner, admin }) => {
+  if (usersInRoom.length === 0) {
+    return null;
+  }
+
+  const activeOwner = usersInRoom.find((user) => sameUser(user, owner));
+  if (activeOwner) {
+    return activeOwner;
+  }
+
+  return usersInRoom.find((user) => sameUser(user, admin)) || usersInRoom[0];
+};
+
 // const lengths = {
 
 // };
@@ -87,6 +103,12 @@ const Room = new mongoose.Schema({
     of: Boolean,
     default: {},
   },
+  // Incremented for every join or departure so a conditional departure cannot
+  // overwrite a reconnect that raced it on another Socket.IO process.
+  membershipRevision: {
+    type: Number,
+    default: 0,
+  },
   admin: { type: mongoose.Schema.Types.ObjectId, ref: 'User' },
   owner: { type: mongoose.Schema.Types.ObjectId, ref: 'User' },
   type: {
@@ -187,6 +209,7 @@ Room.methods.addUser = async function (user, spectating, updateAdmin) {
   }
 
   this.inRoom.set(user.id.toString(), true);
+  this.membershipRevision = (this.membershipRevision || 0) + 1;
 
   if (this.waitingForCount === 0) {
     this.waitingFor.set(user.id.toString(), true);
@@ -210,6 +233,7 @@ Room.methods.addUser = async function (user, spectating, updateAdmin) {
 Room.methods.dropUser = async function (user, updateAdmin) {
   this.inRoom.set(user.id.toString(), false);
   this.waitingFor.set(user.id.toString(), false);
+  this.membershipRevision = (this.membershipRevision || 0) + 1;
 
   await this.updateAdminIfNeeded(updateAdmin);
 
@@ -220,6 +244,70 @@ Room.methods.dropUser = async function (user, updateAdmin) {
   return this.save();
 };
 
+Room.methods.dropUserAtomically = async function (user) {
+  const userKey = user.id.toString();
+  if (!this.inRoom.get(userKey)) {
+    return null;
+  }
+
+  const usersInRoom = this.users.filter((candidate) => (
+    candidate.id.toString() !== userKey
+      && this.inRoom.get(candidate.id.toString())
+  ));
+  const nextAdmin = selectRoomAdmin({
+    usersInRoom,
+    owner: this.owner,
+    admin: this.admin,
+  });
+  const adminChanged = !sameUser(this.admin, nextAdmin);
+  const membershipRevision = this.membershipRevision || 0;
+  const condition = {
+    _id: this._id,
+    [`inRoom.${userKey}`]: true,
+  };
+  if (this.updatedAt) {
+    condition.updatedAt = this.updatedAt;
+  }
+  if (membershipRevision > 0) {
+    condition.membershipRevision = membershipRevision;
+  } else {
+    condition.$or = [
+      { membershipRevision: 0 },
+      { membershipRevision: { $exists: false } },
+    ];
+  }
+
+  const update = {
+    $set: {
+      [`inRoom.${userKey}`]: false,
+      [`waitingFor.${userKey}`]: false,
+      admin: nextAdmin ? nextAdmin._id || nextAdmin : null,
+    },
+    $inc: {
+      membershipRevision: 1,
+    },
+  };
+  if (usersInRoom.length === 0 && this.type === 'normal') {
+    update.$set.expireAt = new Date(Date.now() + STALE_ROOM_LIFETIME_MS);
+  }
+
+  const updatedRoom = await this.constructor.findOneAndUpdate(condition, update, {
+    new: true,
+  }).populate('users').populate('admin').populate('owner');
+  if (!updatedRoom) {
+    return null;
+  }
+
+  await mirrorRoomChanges(updatedRoom, {
+    attempts: [],
+    participantUserIds: [userKey],
+    syncAllParticipants: false,
+    syncRoomOwners: adminChanged,
+  });
+
+  return { room: updatedRoom, adminChanged };
+};
+
 Room.methods.banUser = async function (userId) {
   this.banned.set(userId.toString(), true);
   return this.dropUser({ id: userId });
@@ -314,22 +402,6 @@ Room.methods.edit = async function (options) {
   return this.save();
 };
 
-const sameUser = (left, right) => left && right
-  && String(left.id) === String(right.id);
-
-const selectRoomAdmin = ({ usersInRoom, owner, admin }) => {
-  if (usersInRoom.length === 0) {
-    return null;
-  }
-
-  const activeOwner = usersInRoom.find((user) => sameUser(user, owner));
-  if (activeOwner) {
-    return activeOwner;
-  }
-
-  return usersInRoom.find((user) => sameUser(user, admin)) || usersInRoom[0];
-};
-
 Room.methods.updateAdminIfNeeded = async function (cb) {
   const nextAdmin = selectRoomAdmin(this);
   if (sameUser(this.admin, nextAdmin) || (!this.admin && !nextAdmin)) {

server/models/room.test.js

@@ -4,6 +4,7 @@
 const bcrypt = require('bcrypt');
 const mongoose = require('mongoose');
 const { generateScramble } = require('letscube-scrambles');
+const { mirrorRoomChanges } = require('../postgres/dualWrite');
 const {
   collectPostgresChanges,
   Room,
@@ -13,6 +14,9 @@ const {
 jest.mock('letscube-scrambles', () => ({
   generateScramble: jest.fn(),
 }));
+jest.mock('../postgres/dualWrite', () => ({
+  mirrorRoomChanges: jest.fn().mockResolvedValue(undefined),
+}));
 
 const RoomModel = mongoose.model('RoomPostgresChangesTest', Room);
 
@@ -260,6 +264,54 @@ describe('room security helpers', () => {
     expect(onAdminChange).not.toHaveBeenCalled();
   });
 
+  it('claims a departure with the membership revision compare-and-set', async () => {
+    const owner = { id: 101, _id: 'owner-id' };
+    const nextAdmin = { id: 202, _id: 'next-admin-id' };
+    const updatedRoom = { id: 'room-one' };
+    const query = {
+      populate: jest.fn(() => query),
+      then: (resolve, reject) => Promise.resolve(updatedRoom).then(resolve, reject),
+    };
+    const model = {
+      findOneAndUpdate: jest.fn(() => query),
+    };
+    const updatedAt = new Date('2026-07-13T04:30:00.000Z');
+    const room = {
+      _id: 'room-one',
+      owner,
+      admin: owner,
+      users: [owner, nextAdmin],
+      inRoom: new Map([['101', true], ['202', true]]),
+      type: 'normal',
+      membershipRevision: 7,
+      updatedAt,
+      constructor: model,
+    };
+
+    const departure = await Room.methods.dropUserAtomically.call(room, owner);
+
+    expect(model.findOneAndUpdate).toHaveBeenCalledWith({
+      _id: room._id,
+      'inRoom.101': true,
+      membershipRevision: 7,
+      updatedAt,
+    }, {
+      $set: {
+        'inRoom.101': false,
+        'waitingFor.101': false,
+        admin: nextAdmin._id,
+      },
+      $inc: { membershipRevision: 1 },
+    }, { new: true });
+    expect(departure).toEqual({ room: updatedRoom, adminChanged: true });
+    expect(mirrorRoomChanges).toHaveBeenCalledWith(updatedRoom, {
+      attempts: [],
+      participantUserIds: ['101'],
+      syncAllParticipants: false,
+      syncRoomOwners: true,
+    });
+  });
+
   it('persists and announces the owner reclaiming admin controls', async () => {
     const owner = { id: 101 };
     const room = {

server/socket/namespaces/rooms.js

@@ -127,19 +127,38 @@ module.exports = (io, middlewares) => {
   async function finalizeUserDeparture({
     roomId, userId, connectionId, leaveReason,
   }) {
-    const room = await fetchRoom(roomId);
     const userKey = userId.toString();
 
-    if (!room || !room.inRoom.get(userKey)) {
-      return false;
+    // The compare-and-set in dropUserAtomically is the cross-process claim.
+    // A losing process re-reads so it never overwrites a concurrent rejoin.
+    async function claimDeparture() {
+      const room = await fetchRoom(roomId);
+      if (!room || !room.inRoom.get(userKey)) {
+        return null;
+      }
+
+      if (leaveReason === 'disconnect'
+        && await hasActiveSocketsForUserRoom(userId, roomId)) {
+        return null;
+      }
+
+      const departure = await room.dropUserAtomically({ id: userId });
+      if (!departure) {
+        return claimDeparture();
+      }
+
+      return departure;
     }
 
-    if (leaveReason === 'disconnect'
-      && await hasActiveSocketsForUserRoom(userId, roomId)) {
+    const departure = await claimDeparture();
+    if (!departure) {
       return false;
     }
 
-    const updatedRoom = await room.dropUser({ id: userId }, sendAdminUpdate);
+    const { room: updatedRoom, adminChanged } = departure;
+    if (adminChanged) {
+      sendAdminUpdate(updatedRoom);
+    }
 
     await metrics.endRoomVisit({
       room: updatedRoom,

server/socket/namespaces/rooms.test.js

@@ -363,11 +363,12 @@ describe('room namespace departures and moderation', () => {
       inRoom: new Map([['101', true]]),
       doneWithScramble: jest.fn().mockReturnValue(false),
     });
-    room.dropUser = jest.fn(async (user, onAdminChange) => {
-      room.inRoom.set(String(user.id), false);
-      room.admin = null;
-      onAdminChange(room);
-      return room;
+    room.dropUserAtomically = jest.fn().mockResolvedValue({
+      room: {
+        ...room,
+        admin: null,
+      },
+      adminChanged: true,
     });
     const connect = setup(new Map([[room._id, room]]));
     await connect(makeSocket({ id: 'spectator-socket', session: {} }));
@@ -384,6 +385,55 @@ describe('room namespace departures and moderation', () => {
     expect(connect.roomChannel.emit).toHaveBeenCalledWith(Protocol.USER_LEFT, 101);
   });
 
+  it('emits metrics and leave events only for the process that claims a departure', async () => {
+    const room = makeRoom({
+      private: false,
+      password: null,
+      inRoom: new Map([['101', true]]),
+    });
+    const updatedRoom = {
+      ...room,
+      inRoom: new Map([['101', false]]),
+      usersLength: 0,
+      doneWithScramble: jest.fn().mockReturnValue(false),
+    };
+    let claimed = false;
+    room.dropUserAtomically = jest.fn(async () => {
+      if (claimed) {
+        return null;
+      }
+      claimed = true;
+      return { room: updatedRoom, adminChanged: false };
+    });
+    const connect = setup(new Map([[room._id, room]]));
+    const persistedRoom = {
+      ...room,
+      inRoom: new Map([['101', false]]),
+    };
+    Room.findById.mockImplementation(() => queryResult(claimed ? persistedRoom : room));
+    const reconnectGrace = createReconnectGrace.mock.results[0].value;
+
+    await Promise.all([
+      reconnectGrace.finalizeDeparture({
+        roomId: room._id,
+        userId: 101,
+        connectionId: 'socket-on-process-a',
+        leaveReason: 'explicit',
+      }),
+      reconnectGrace.finalizeDeparture({
+        roomId: room._id,
+        userId: 101,
+        connectionId: 'socket-on-process-b',
+        leaveReason: 'explicit',
+      }),
+    ]);
+
+    expect(room.dropUserAtomically).toHaveBeenCalledTimes(2);
+    expect(metrics.endRoomVisit).toHaveBeenCalledTimes(1);
+    expect(connect.roomChannel.emit).toHaveBeenCalledTimes(1);
+    expect(connect.roomChannel.emit).toHaveBeenCalledWith(Protocol.USER_LEFT, 101);
+  });
+
   it('finalizes one departure when two tabs explicitly leave together', async () => {
     const room = makeRoom({
       private: false,