fix(collab): serialize fan-out per socket with a flag, not by pre-claiming rev

The race this PR identified is real. With the guard removed, three concurrent
updatePadClients() runs deliver [1,1,2,1,2,3,2,3,3] to one client — every
revision three times (new spec, padFanOutSerialization.ts).

But claiming the range by writing `sessioninfo.rev = headRev` before the awaits
breaks a guarantee that rev carries elsewhere. handleUserChanges asserts
`thisSession.rev === r` right before emitting ACCEPT_COMMIT, with the comment
"make sure we have already sent any previous ACCEPT_COMMIT and NEW_CHANGES
messages". Pre-claiming satisfies that assert while the NEW_CHANGES for those
revisions are still queued behind `await pad.getRevision()`, so ACCEPT_COMMIT can
overtake them — the exact out-of-order delivery the assert exists to prevent.
The rollback path has the same problem in reverse: a mid-batch emit failure
rewinds to startRev and re-sends revisions the client already applied, where the
per-revision commit only retried the one that failed.

So: keep the loop as it was, add `sessioninfo.fanOutInFlight`. At most one fan-out
per socket, rev keeps meaning "last revision actually sent", retry semantics
unchanged, and a skipped run loses nothing because the holder re-reads
getHeadRevisionNumber() each iteration.

Change B (NEW_CHANGES_BATCH) is dropped. It never fired in the dive by the PR's
own account, it changes the wire format behind a flag that needs client
coordination, and it does not belong in a correctness fix. Worth its own PR if
the packet-count argument is ever measured.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: 2 people

settings.json.template

@@ -821,18 +821,6 @@
    */
   "loadTest": false,
 
-  /*
-   * Pack multiple NEW_CHANGES revisions into a single NEW_CHANGES_BATCH emit
-   * per recipient when a fan-out has more than one revision to catch up
-   * (#7756 lever 3b). Reduces engine.io packet count under high pad
-   * concurrency, especially on the WebSocket transport.
-   *
-   * WARNING: enabling this requires all connected clients to understand
-   * NEW_CHANGES_BATCH. Old clients will silently fail to apply batched
-   * revisions. Coordinate the rollout.
-   */
-  "newChangesBatch": false,
-
   /**
   * Disable dump of objects preventing a clean exit
   */

src/node/handler/NewChangesPacker.ts

@@ -49,7 +49,6 @@ const hooks = require('../../static/js/pluginfw/hooks');
 const stats = require('../stats')
 const assert = require('assert').strict;
 import {recordChangesetApply, recordSocketEmit} from '../prom-instruments';
-import {buildNewChangesEmits, type NewChangesItem} from './NewChangesPacker';
 import {RateLimiterMemory} from 'rate-limiter-flexible';
 import {ChangesetRequest, PadUserInfo, SocketClientRequest} from "../types/SocketClientRequest";
 import {APool, AText, PadAuthor, PadType} from "../types/PadType";
@@ -1053,76 +1052,69 @@ exports.updatePadClients = async (pad: PadType) => {
   // but benefit of reusing cached revision object is HUGE
   const revCache:MapArrayType<any> = {};
 
-  // When `settings.newChangesBatch` is true and a recipient is more than one
-  // revision behind, pack the queued revisions into a single NEW_CHANGES_BATCH
-  // emit per recipient. The engine.io WebSocket transport sends one frame per
-  // packet (the polling transport already batches at the HTTP-response layer),
-  // so reducing the packet count translates directly into fewer system calls
-  // on the server and fewer onmessage callbacks on the client.
-  const batchEnabled = settings.newChangesBatch === true;
-
   await Promise.all(roomSockets.map(async (socket) => {
     const sessioninfo = sessioninfos[socket.id];
     // The user might have disconnected since _getRoomSockets() was called.
     if (sessioninfo == null) return;
 
-    // Snapshot the local state so a concurrent updatePadClients() can't make
-    // us double-emit. We hold the "I'm responsible for revs (startRev,
-    // headRev]" claim by reading sessioninfo.rev once and overwriting it
-    // before any await. A second invocation arriving mid-loop will see the
-    // bumped rev and skip those revisions; if our emit fails the catch
-    // below rolls sessioninfo.rev back so they aren't lost.
-    const startRev = sessioninfo.rev;
-    const headRev = pad.getHeadRevisionNumber();
-    if (startRev >= headRev) return;
-    const startTime = sessioninfo.time;
-    // Claim the range immediately so concurrent runs skip it.
-    sessioninfo.rev = headRev;
-
-    // Collect all queued revisions for this socket.
-    const pending: Array<{
-      newRev: number;
-      changeset: string;
-      apool: unknown;
-      author: string;
-      currentTime: number;
-      timeDelta: number;
-    }> = [];
-
+    // 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 {
-      let previousTime = startTime;
-      for (let r = startRev + 1; r <= headRev; r++) {
+      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;
-        const forWire = prepareForWire(revChangeset, pad.pool);
-        pending.push({
-          newRev: r,
-          changeset: forWire.translated,
-          apool: forWire.pool,
-          author,
-          currentTime,
-          timeDelta: currentTime - previousTime,
-        });
-        previousTime = currentTime;
-      }
 
-      for (const emit of buildNewChangesEmits(pending, batchEnabled)) {
-        socket.emit('message', emit);
-        recordSocketEmit(emit.data.type);
+        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;
       }
-      // Only after the wire send succeeds do we commit the new time.
-      sessioninfo.time = previousTime;
-    } catch (err: any) {
-      // Roll back the claim so the next updatePadClients retries these revs.
-      // Only set rev back if no one else has advanced past us in the meantime.
-      if (sessioninfo.rev === headRev) sessioninfo.rev = startRev;
-      messageLogger.error(`Failed to notify user of new revision: ${err.stack || err}`);
+    } 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/node/handler/PadMessageHandler.ts

@@ -40,7 +40,6 @@ export const padUsersGauge = new client.Gauge({
 // state.
 const KNOWN_TYPES = new Set([
   'NEW_CHANGES',
-  'NEW_CHANGES_BATCH',
   'ACCEPT_COMMIT',
   'CHAT_MESSAGE',
   'CLIENT_VARS',

src/node/prom-instruments.ts

@@ -292,7 +292,6 @@ export type SettingsType = {
   automaticReconnectionTimeout: number,
   loadTest: boolean,
   scalingDiveMetrics: boolean,
-  newChangesBatch: boolean,
   dumpOnUncleanExit: boolean,
   indentationOnNewLine: boolean,
   logconfig: any | null,
@@ -732,21 +731,6 @@ const settings: SettingsType = {
    * production deployments aren't paying for instrumentation they don't use.
    */
   scalingDiveMetrics: false,
-  /**
-   * Pack multiple NEW_CHANGES revisions into a single NEW_CHANGES_BATCH emit
-   * per recipient when a fan-out has more than one revision to catch up
-   * (#7756 lever 3b). Reduces engine.io packet count under high pad
-   * concurrency, especially on the WebSocket transport, which sends one
-   * frame per packet (the polling transport already batches naturally).
-   *
-   * Requires clients to recognise the NEW_CHANGES_BATCH message type. New
-   * clients are forward-compatible (they handle both NEW_CHANGES and
-   * NEW_CHANGES_BATCH). Old clients connecting to a server with this enabled
-   * would fail to apply batched revisions, so coordinate the rollout.
-   *
-   * 0/false (default) preserves legacy per-revision emit behaviour.
-   */
-  newChangesBatch: false,
   /**
    * Disable dump of objects preventing a clean exit
    */

src/node/utils/Settings.ts

@@ -493,23 +493,17 @@ const loadBroadcastJS = (socket, sendSocketMsg, fireWhenAllScriptsAreLoaded, Bro
       if (obj.type === 'COLLABROOM') {
         obj = obj.data;
 
-        if (obj.type === 'NEW_CHANGES' || obj.type === 'NEW_CHANGES_BATCH') {
-          // NEW_CHANGES_BATCH (#7756 lever 3b) carries an array of revisions
-          // in one emit. Each revision has the same shape as the legacy
-          // single-rev message; apply in order.
-          const changes = obj.type === 'NEW_CHANGES_BATCH' ? obj.changes : [obj];
-          for (const change of changes) {
-            const changeset = moveOpsToNewPool(
-                change.changeset, (new AttribPool()).fromJsonable(change.apool), padContents.apool);
-
-            let changesetBack = inverse(
-                change.changeset, padContents.currentLines, padContents.alines, padContents.apool);
-
-            changesetBack = moveOpsToNewPool(
-                changesetBack, (new AttribPool()).fromJsonable(change.apool), padContents.apool);
-
-            loadedNewChangeset(changeset, changesetBack, change.newRev - 1, change.timeDelta);
-          }
+        if (obj.type === 'NEW_CHANGES') {
+          const changeset = moveOpsToNewPool(
+              obj.changeset, (new AttribPool()).fromJsonable(obj.apool), padContents.apool);
+
+          let changesetBack = inverse(
+              obj.changeset, padContents.currentLines, padContents.alines, padContents.apool);
+
+          changesetBack = moveOpsToNewPool(
+              changesetBack, (new AttribPool()).fromJsonable(obj.apool), padContents.apool);
+
+          loadedNewChangeset(changeset, changesetBack, obj.newRev - 1, obj.timeDelta);
         } else if (obj.type === 'NEW_AUTHORDATA') {
           const authorMap = {};
           authorMap[obj.author] = obj.data;

src/static/js/broadcast.ts

@@ -206,12 +206,7 @@ const getCollabClient = (ace2editor, serverVars, initialUserInfo, options, _pad)
     if (wrapper.type !== 'COLLABROOM' && wrapper.type !== 'CUSTOM') return;
     const msg = wrapper.data;
 
-    if (msg.type === 'NEW_CHANGES' || msg.type === 'NEW_CHANGES_BATCH') {
-      // NEW_CHANGES_BATCH (added in #7756 lever 3b) carries an array of
-      // revisions in one emit. Each revision has the same shape as the
-      // legacy single-rev message, so we normalise to a list and apply in
-      // order, sharing the same compose-safety await.
-      const changes = msg.type === 'NEW_CHANGES_BATCH' ? msg.changes : [msg];
+    if (msg.type === 'NEW_CHANGES') {
       serverMessageTaskQueue.enqueue(async () => {
         // Avoid updating the DOM while the user is composing a character. Notes about this `await`:
         //   * `await null;` is equivalent to `await Promise.resolve(null);`, so if the user is not
@@ -221,16 +216,14 @@ const getCollabClient = (ace2editor, serverVars, initialUserInfo, options, _pad)
         //     possible, that the chances are so small or the consequences so minor that it's not
         //     worth addressing).
         await editor.getInInternationalComposition();
-        for (const change of changes) {
-          const {newRev, changeset, author = '', apool} = change;
-          if (newRev !== (rev + 1)) {
-            window.console.warn(`bad message revision on ${msg.type}: ${newRev} not ${rev + 1}`);
-            // setChannelState("DISCONNECTED", "badmessage_newchanges");
-            return;
-          }
-          rev = newRev;
-          editor.applyChangesToBase(changeset, author, apool);
+        const {newRev, changeset, author = '', apool} = msg;
+        if (newRev !== (rev + 1)) {
+          window.console.warn(`bad message revision on NEW_CHANGES: ${newRev} not ${rev + 1}`);
+          // setChannelState("DISCONNECTED", "badmessage_newchanges");
+          return;
         }
+        rev = newRev;
+        editor.applyChangesToBase(changeset, author, apool);
       });
     } else if (msg.type === 'ACCEPT_COMMIT') {
       serverMessageTaskQueue.enqueue(() => {

src/static/js/collab_client.ts

@@ -128,20 +128,6 @@ export type ClientNewChanges = {
   payload?: ClientNewChanges
 }
 
-export type NewChangesItem = {
-  apool: AttributePool,
-  author: string,
-  changeset: string,
-  currentTime: number,
-  newRev: number,
-  timeDelta: number,
-}
-
-export type ClientNewChangesBatch = {
-  type: 'NEW_CHANGES_BATCH',
-  changes: NewChangesItem[],
-}
-
 export type ClientAcceptCommitMessage = {
   type: 'ACCEPT_COMMIT'
   newRev: number