fix(new-changes-batch): address Qodo review on #7768

Two issues raised on the first push:

1. **Rev advanced before send (Bug · Correctness).** The previous
   diff advanced sessioninfo.rev/time inside the collect loop,
   before any emit ran. A concurrent updatePadClients() could then
   see the bumped rev and skip those revisions, and if the emit
   threw later, the skipped revs were lost forever. The client
   enforces strict newRev===rev+1 and silently stops applying on
   mismatch — net effect was a possible pad desync under
   concurrent fan-outs.

   Fix: snapshot startRev/startTime once, claim the
   (startRev, headRev] range by setting sessioninfo.rev = headRev
   immediately (so a concurrent run skips it), build the pending
   list against the local startTime, then emit. If the emit
   throws, roll sessioninfo.rev back to startRev so the next
   fan-out retries. Time is only committed after a successful
   send.

2. **Test re-implemented the decision (Rule violation ·
   Reliability).** The original test re-implemented the
   NEW_CHANGES vs NEW_CHANGES_BATCH switch locally instead of
   exercising the production code. Removing the production logic
   would have left the test green.

   Fix: extract the pure wire-format decision into
   src/node/handler/NewChangesPacker.ts (no DB / pad dependency,
   so the test can import it directly under vitest), and rewrite
   the test to assert against the exported `buildNewChangesEmits`
   function from that module. PadMessageHandler now calls the
   same function; deleting it would fail the test.

9/9 tests across new-changes-batch + prom-instruments.

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

src/node/handler/NewChangesPacker.ts

@@ -0,0 +1,41 @@
+// Wire-format decision for NEW_CHANGES vs NEW_CHANGES_BATCH (#7756 lever 3b).
+//
+// Lives in its own tiny module rather than inside PadMessageHandler so the
+// pure decision can be unit-tested without standing up the full pad / DB /
+// socket.io stack. PadMessageHandler.updatePadClients calls this function
+// once per recipient with the queued revisions for that recipient.
+
+export type NewChangesItem = {
+  newRev: number;
+  changeset: string;
+  apool: unknown;
+  author: string;
+  currentTime: number;
+  timeDelta: number;
+};
+
+export type NewChangesEmit =
+  | {type: 'COLLABROOM'; data: {type: 'NEW_CHANGES'} & NewChangesItem}
+  | {type: 'COLLABROOM'; data: {type: 'NEW_CHANGES_BATCH'; changes: NewChangesItem[]}};
+
+/**
+ * Decide what to put on the wire for one recipient.
+ * - No queued revisions: nothing.
+ * - Batching disabled, or exactly one rev: emit one NEW_CHANGES per rev
+ *   (legacy behaviour; preserves bytes-on-wire for the steady state).
+ * - Batching enabled and multiple revs: emit one NEW_CHANGES_BATCH with
+ *   the array of revisions.
+ */
+export const buildNewChangesEmits = (
+  pending: NewChangesItem[],
+  batchEnabled: boolean,
+): NewChangesEmit[] => {
+  if (pending.length === 0) return [];
+  if (batchEnabled && pending.length > 1) {
+    return [{type: 'COLLABROOM', data: {type: 'NEW_CHANGES_BATCH', changes: pending}}];
+  }
+  return pending.map((change) => ({
+    type: 'COLLABROOM',
+    data: {type: 'NEW_CHANGES', ...change},
+  } as NewChangesEmit));
+};

src/node/handler/PadMessageHandler.ts

@@ -48,6 +48,7 @@ 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";
@@ -977,6 +978,19 @@ exports.updatePadClients = async (pad: PadType) => {
     // 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;
@@ -987,50 +1001,39 @@ exports.updatePadClients = async (pad: PadType) => {
       timeDelta: number;
     }> = [];
 
-    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 - sessioninfo.time,
-      });
-      sessioninfo.time = currentTime;
-      sessioninfo.rev = r;
-    }
-
-    if (pending.length === 0) return;
-
     try {
-      if (batchEnabled && pending.length > 1) {
-        socket.emit('message', {
-          type: 'COLLABROOM',
-          data: {type: 'NEW_CHANGES_BATCH', changes: pending},
-        });
-        recordSocketEmit('NEW_CHANGES_BATCH');
-      } else {
-        for (const change of pending) {
-          socket.emit('message', {
-            type: 'COLLABROOM',
-            data: {type: 'NEW_CHANGES', ...change},
-          });
-          recordSocketEmit('NEW_CHANGES');
+      let previousTime = startTime;
+      for (let r = startRev + 1; r <= headRev; r++) {
+        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);
+      }
+      // 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}`);
     }
   }));

src/tests/backend-new/specs/new-changes-batch.test.ts

@@ -1,73 +1,48 @@
-// Unit coverage for the NEW_CHANGES_BATCH server-side packing
-// (#7756 lever 3b). Server-side concern only — verifies that the
-// pad fan-out emits one batch per recipient when multiple revs queue
-// up and the feature flag is on, and falls back to per-rev emits
-// otherwise. Client-side coverage lives in the existing Playwright
-// flow tests; this test pins the wire-format decision.
+// Regression test for the NEW_CHANGES_BATCH wire-format decision
+// (#7756 lever 3b). Imports the real implementation from
+// PadMessageHandler so removing or breaking the production batching
+// logic fails this test.
 
 import {describe, it, expect, beforeEach, afterEach} from 'vitest';
 import settings from '../../../node/utils/Settings';
+import {buildNewChangesEmits, type NewChangesItem} from '../../../node/handler/NewChangesPacker';
 
 const ORIGINAL_FLAG = settings.newChangesBatch;
 
 beforeEach(() => { settings.newChangesBatch = false; });
 afterEach(() => { settings.newChangesBatch = ORIGINAL_FLAG; });
 
-// The decision the new code makes is small and pure: given a `pending`
-// array of N >= 1 revisions and the feature flag, emit one
-// NEW_CHANGES_BATCH (if N > 1 and flag on) or N NEW_CHANGES messages.
-// Re-implement the decision here so the test doesn't have to stand up
-// the full pad/DB stack — and pin it against the actual implementation
-// via a comment in PadMessageHandler.
-
-type Pending = {newRev: number; changeset: string; apool: unknown;
-                author: string; currentTime: number; timeDelta: number};
-type Emit = {type: 'COLLABROOM'; data: any};
-
-const decideEmits = (pending: Pending[], batchEnabled: boolean): Emit[] => {
-  if (pending.length === 0) return [];
-  if (batchEnabled && pending.length > 1) {
-    return [{type: 'COLLABROOM', data: {type: 'NEW_CHANGES_BATCH', changes: pending}}];
-  }
-  return pending.map((change) => ({
-    type: 'COLLABROOM',
-    data: {type: 'NEW_CHANGES', ...change},
-  }));
-};
-
-const fakePending = (n: number): Pending[] =>
+const fakePending = (n: number): NewChangesItem[] =>
   Array.from({length: n}, (_, i) => ({
     newRev: i + 1, changeset: `=${i}`, apool: {}, author: 'a.1',
     currentTime: 1_000 * (i + 1), timeDelta: 1_000,
   }));
 
-describe('NEW_CHANGES_BATCH emit decision', () => {
-  it('with flag OFF, sends one NEW_CHANGES per rev regardless of count', () => {
-    settings.newChangesBatch = false;
-    const emits = decideEmits(fakePending(5), settings.newChangesBatch);
+describe('buildNewChangesEmits', () => {
+  it('flag OFF: one NEW_CHANGES per rev regardless of count', () => {
+    const emits = buildNewChangesEmits(fakePending(5), false);
     expect(emits).toHaveLength(5);
     expect(emits.every((e) => e.data.type === 'NEW_CHANGES')).toBe(true);
   });
 
-  it('with flag ON and one queued rev, still sends NEW_CHANGES (no batch overhead)', () => {
-    settings.newChangesBatch = true;
-    const emits = decideEmits(fakePending(1), settings.newChangesBatch);
+  it('flag ON, single rev: still NEW_CHANGES (no batch overhead for the steady state)', () => {
+    const emits = buildNewChangesEmits(fakePending(1), true);
     expect(emits).toHaveLength(1);
     expect(emits[0]!.data.type).toBe('NEW_CHANGES');
   });
 
-  it('with flag ON and multiple queued revs, sends one NEW_CHANGES_BATCH', () => {
-    settings.newChangesBatch = true;
-    const emits = decideEmits(fakePending(5), settings.newChangesBatch);
+  it('flag ON, multiple revs: a single NEW_CHANGES_BATCH carrying all of them', () => {
+    const emits = buildNewChangesEmits(fakePending(5), true);
     expect(emits).toHaveLength(1);
     expect(emits[0]!.data.type).toBe('NEW_CHANGES_BATCH');
-    expect(emits[0]!.data.changes).toHaveLength(5);
-    expect(emits[0]!.data.changes[0]!.newRev).toBe(1);
-    expect(emits[0]!.data.changes[4]!.newRev).toBe(5);
+    const batch = emits[0]!.data as {type: 'NEW_CHANGES_BATCH'; changes: NewChangesItem[]};
+    expect(batch.changes).toHaveLength(5);
+    expect(batch.changes[0]!.newRev).toBe(1);
+    expect(batch.changes[4]!.newRev).toBe(5);
   });
 
   it('empty pending list emits nothing', () => {
-    settings.newChangesBatch = true;
-    expect(decideEmits([], settings.newChangesBatch)).toEqual([]);
+    expect(buildNewChangesEmits([], true)).toEqual([]);
+    expect(buildNewChangesEmits([], false)).toEqual([]);
   });
 });