feat(scaling): serialize per-socket fan-out + NEW_CHANGES_BATCH (#7756 lever 3)

src/node/handler/PadMessageHandler.ts

@@ -1056,40 +1056,64 @@ exports.updatePadClients = async (pad: PadType) => {
     // The user might have disconnected since _getRoomSockets() was called.
     if (sessioninfo == null) return;
 
-    while (sessioninfo.rev < pad.getHeadRevisionNumber()) {
-      const r = sessioninfo.rev + 1;
-      let revision = revCache[r];
-      if (!revision) {
-        revision = await pad.getRevision(r);
-        revCache[r] = revision;
-      }
-
-      const author = revision.meta.author;
-      const revChangeset = revision.changeset;
-      const currentTime = revision.meta.timestamp;
+    // One fan-out per socket at a time. Without this, two concurrent
+    // updatePadClients() runs both read `sessioninfo.rev`, both await
+    // pad.getRevision(), and both emit the same revision to the same client
+    // (#7756 lever 3): the loop's read-await-write is not atomic.
+    //
+    // The claim is a separate flag on purpose. `sessioninfo.rev` has to keep
+    // meaning "last revision actually put on the wire" — handleUserChanges
+    // asserts `thisSession.rev === r` before emitting ACCEPT_COMMIT precisely to
+    // guarantee the client sees NEW_CHANGES and ACCEPT_COMMIT in order. Bumping
+    // rev up front to claim a range would satisfy that assert while the
+    // NEW_CHANGES messages are still queued behind an await, which is the
+    // out-of-order delivery the assert exists to prevent.
+    //
+    // A skipped run loses nothing: the run that holds the flag re-reads
+    // getHeadRevisionNumber() every iteration, so it also ships whatever arrived
+    // in the meantime.
+    if (sessioninfo.fanOutInFlight) return;
+    sessioninfo.fanOutInFlight = true;
+    try {
+      while (sessioninfo.rev < pad.getHeadRevisionNumber()) {
+        const r = sessioninfo.rev + 1;
+        let revision = revCache[r];
+        if (!revision) {
+          revision = await pad.getRevision(r);
+          revCache[r] = revision;
+        }
 
-      const forWire = prepareForWire(revChangeset, pad.pool);
-      const msg = {
-        type: 'COLLABROOM',
-        data: {
-          type: 'NEW_CHANGES',
-          newRev: r,
-          changeset: forWire.translated,
-          apool: forWire.pool,
-          author,
-          currentTime,
-          timeDelta: currentTime - sessioninfo.time,
-        },
-      };
-      try {
-        socket.emit('message', msg);
-        recordSocketEmit('NEW_CHANGES');
-      } catch (err:any) {
-        messageLogger.error(`Failed to notify user of new revision: ${err.stack || err}`);
-        return;
+        const author = revision.meta.author;
+        const revChangeset = revision.changeset;
+        const currentTime = revision.meta.timestamp;
+
+        const forWire = prepareForWire(revChangeset, pad.pool);
+        const msg = {
+          type: 'COLLABROOM',
+          data: {
+            type: 'NEW_CHANGES',
+            newRev: r,
+            changeset: forWire.translated,
+            apool: forWire.pool,
+            author,
+            currentTime,
+            timeDelta: currentTime - sessioninfo.time,
+          },
+        };
+        try {
+          socket.emit('message', msg);
+          recordSocketEmit('NEW_CHANGES');
+        } catch (err:any) {
+          messageLogger.error(`Failed to notify user of new revision: ${err.stack || err}`);
+          return;
+        }
+        sessioninfo.time = currentTime;
+        sessioninfo.rev = r;
       }
-      sessioninfo.time = currentTime;
-      sessioninfo.rev = r;
+    } finally {
+      // Cleared even on the `return` inside the emit-failure path above, so a
+      // failed send doesn't wedge the socket's fan-out for good.
+      sessioninfo.fanOutInFlight = false;
     }
   }));
 };

src/tests/backend/specs/padFanOutSerialization.ts

@@ -0,0 +1,101 @@
+'use strict';
+
+/**
+ * updatePadClients() fan-out must not double-send a revision to a client
+ * (#7756 lever 3).
+ *
+ * The per-socket loop reads `sessioninfo.rev`, awaits `pad.getRevision()`, then
+ * writes `sessioninfo.rev` back. That read-await-write is not atomic, so two
+ * concurrent fan-outs for the same pad both start from the same rev and both emit
+ * it. The client applies the changeset twice and its revision numbering diverges
+ * from the server's.
+ *
+ * Guarded by `sessioninfo.fanOutInFlight`, deliberately NOT by pre-claiming
+ * `sessioninfo.rev`: handleUserChanges asserts `thisSession.rev === r` to keep
+ * NEW_CHANGES and ACCEPT_COMMIT ordered on the wire, and that assert is only
+ * meaningful while `rev` means "already sent".
+ */
+
+import {PadType} from '../../../node/types/PadType';
+
+const assert = require('assert').strict;
+const common = require('../common');
+const padManager = require('../../../node/db/PadManager');
+const padMessageHandler = require('../../../node/handler/PadMessageHandler');
+
+let agent: any;
+
+describe(__filename, function () {
+  this.timeout(30000);
+
+  let pad: PadType;
+  let padId: string;
+  let socket: any;
+
+  before(async function () {
+    agent = await common.init();
+  });
+
+  beforeEach(async function () {
+    padId = common.randomString();
+    pad = await padManager.getPad(padId, 'x\n');
+    const res = await agent.get(`/p/${padId}`).expect(200);
+    socket = await common.connect(res);
+    const {type} = await common.handshake(socket, padId);
+    assert.equal(type, 'CLIENT_VARS');
+  });
+
+  afterEach(async function () {
+    if (socket != null) socket.close();
+    socket = null;
+    await padManager.getPad(padId).then((p: PadType) => p.remove());
+  });
+
+  const collectNewChanges = (ms: number) => {
+    const revs: number[] = [];
+    const onMessage = (msg: any) => {
+      if (msg?.type === 'COLLABROOM' && msg?.data?.type === 'NEW_CHANGES') {
+        revs.push(msg.data.newRev);
+      }
+    };
+    socket.on('message', onMessage);
+    return new Promise<number[]>((resolve) => setTimeout(() => {
+      socket.off('message', onMessage);
+      resolve(revs);
+    }, ms));
+  };
+
+  it('sends each revision exactly once when fan-outs overlap', async function () {
+    const collected = collectNewChanges(2000);
+    // Append behind the client's back so the pad has a backlog to fan out, then
+    // race two fan-outs. Without the guard both runs start at the same rev.
+    for (const text of ['a\n', 'b\n', 'c\n']) await pad.setText(text);
+    const head = pad.getHeadRevisionNumber();
+    await Promise.all([
+      padMessageHandler.updatePadClients(pad),
+      padMessageHandler.updatePadClients(pad),
+      padMessageHandler.updatePadClients(pad),
+    ]);
+
+    const revs = await collected;
+    assert.deepEqual([...new Set(revs)], revs,
+        `every revision must be sent once, got ${JSON.stringify(revs)}`);
+    assert.deepEqual(revs, [...revs].sort((a, b) => a - b),
+        `revisions must arrive in order, got ${JSON.stringify(revs)}`);
+    assert.equal(revs[revs.length - 1], head, 'the client is caught up to head');
+  });
+
+  it('releases the in-flight flag so later fan-outs still deliver', async function () {
+    // Collect across both fan-outs: a flag that is never cleared would deliver
+    // the first revision and then go silent.
+    const collected = collectNewChanges(2000);
+    await pad.setText('first\n');
+    await padMessageHandler.updatePadClients(pad);
+    await pad.setText('second\n');
+    const head = pad.getHeadRevisionNumber();
+    await padMessageHandler.updatePadClients(pad);
+    const revs = await collected;
+    assert.deepEqual(revs, [head - 1, head],
+        `both revisions must be delivered, got ${JSON.stringify(revs)}`);
+  });
+});