perf(pad): prefetch rebase revisions in one Promise.all (#7756)

The rebase loop in handleUserChanges and the catch-up loop in
updatePadClients each did `await pad.getRevision(r)` per intermediate
revision. Under steady-state CPU pressure (the regime the scaling
dive sweeps run in), every await is an event-loop yield that queues
the continuation behind other work. At the cliff that translates
into measured apply_mean of ~40-50 ms per commit even though the
synchronous work is ~5 ms.

Both loops now snapshot headRev once and prefetch the full range
in a single Promise.all. The actual rebase / fan-out iteration runs
synchronously over the prefetched revisions: N event-loop yields
collapse to 1.

In updatePadClients the prefetch covers the SUPERSET range (from
the laggiest recipient's next rev up to headRev) so the per-socket
loop reads from the local revCache without further awaits. The
fallback `await pad.getRevision(r)` is kept as defensive code in
case the head somehow advances mid-loop.

Snapshotting headRev once also makes the existing
`assert([r, r + 1].includes(newRev))` in handleUserChanges more
stable — the assertion was racy if other writers landed commits
during the previous serial-await loop.

Tests: rebase-prefetch.test.ts pins the helper's parallel-fetch
shape and result ordering.

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

src/node/handler/PadMessageHandler.ts

@@ -883,14 +883,30 @@ const handleUserChanges = async (socket:any, message: {
     let rebasedChangeset = moveOpsToNewPool(changeset, wireApool, pad.pool);
 
     // ex. applyUserChanges
-    let r = baseRev;
-
+    //
     // The client's changeset might not be based on the latest revision,
-    // since other clients are sending changes at the same time.
-    // Update the changeset so that it can be applied to the latest revision.
-    while (r < pad.getHeadRevisionNumber()) {
+    // since other clients are sending changes at the same time. Rebase
+    // through every intermediate revision so the changeset applies cleanly
+    // against the current head.
+    //
+    // Prefetching the whole range in one Promise.all collapses N
+    // event-loop yields (one `await pad.getRevision(r)` per intermediate
+    // revision) into 1, materially reducing apply_mean under steady-state
+    // CPU pressure (#7756). Snapshot the head once so we rebase to a
+    // stable target — any commits landing after this point get caught up
+    // by the next handleUserChanges invocation, matching the existing
+    // assertion below (newRev ∈ {r, r+1}).
+    const rebaseTargetHead = pad.getHeadRevisionNumber();
+    const rebaseRange: number[] = [];
+    for (let i = baseRev + 1; i <= rebaseTargetHead; i++) rebaseRange.push(i);
+    const rebaseRevs = rebaseRange.length === 0
+      ? []
+      : await Promise.all(rebaseRange.map((i) => pad.getRevision(i)));
+
+    let r = baseRev;
+    for (const revision of rebaseRevs) {
       r++;
-      const {changeset: c, meta: {author: authorId}} = await pad.getRevision(r);
+      const {changeset: c, meta: {author: authorId}} = revision;
       if (changeset === c && thisSession.author === authorId) {
         // Assume this is a retransmission of an already applied changeset.
         rebasedChangeset = identity(unpack(changeset).oldLen);
@@ -952,27 +968,39 @@ exports.updatePadClients = async (pad: PadType) => {
   const roomSockets = _getRoomSockets(pad.id);
   if (roomSockets.length === 0) return;
 
-  // since all clients usually get the same set of changesets, store them in local cache
-  // to remove unnecessary roundtrip to the datalayer
-  // NB: note below possibly now accommodated via the change to promises/async
-  // TODO: in REAL world, if we're working without datalayer cache,
-  // all requests to revisions will be fired
-  // BEFORE first result will be landed to our cache object.
-  // The solution is to replace parallel processing
-  // via async.forEach with sequential for() loop. There is no real
-  // benefits of running this in parallel,
-  // but benefit of reusing cached revision object is HUGE
-  const revCache:MapArrayType<any> = {};
+  // Find the range of revisions any recipient could possibly need to catch
+  // up to. The per-socket while loop further down won't necessarily walk
+  // the whole range for every recipient, but prefetching the SUPERSET in
+  // one Promise.all collapses N event-loop yields (#7756). At the cliff
+  // step in dive runs this is the difference between every per-rev
+  // `await pad.getRevision(r)` queuing behind other work and the rebase
+  // resolving in microseconds.
+  const headRev = pad.getHeadRevisionNumber();
+  let minRevNeeded = headRev + 1;
+  for (const socket of roomSockets) {
+    const sinfo = sessioninfos[socket.id];
+    if (sinfo == null) continue;
+    if (sinfo.rev + 1 < minRevNeeded) minRevNeeded = sinfo.rev + 1;
+  }
+  const revCache: MapArrayType<any> = {};
+  if (minRevNeeded <= headRev) {
+    const ids: number[] = [];
+    for (let r = minRevNeeded; r <= headRev; r++) ids.push(r);
+    const fetched = await Promise.all(ids.map((r) => pad.getRevision(r as unknown as string)));
+    fetched.forEach((rev, i) => { revCache[ids[i]!] = rev; });
+  }
 
   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;
 
-    while (sessioninfo.rev < pad.getHeadRevisionNumber()) {
+    while (sessioninfo.rev < headRev) {
       const r = sessioninfo.rev + 1;
       let revision = revCache[r];
       if (!revision) {
+        // Should never happen given the prefetch above, but stay robust if
+        // headRev advances mid-loop (shouldn't, since we snapshotted).
         revision = await pad.getRevision(r);
         revCache[r] = revision;
       }

src/tests/backend-new/specs/rebase-prefetch.test.ts

@@ -0,0 +1,66 @@
+// Smoke test for the rebase-loop prefetch optimisation in
+// PadMessageHandler.handleUserChanges (#7756). Exercises the pure
+// "given a baseRev and a head, prefetch revs in one Promise.all" decision
+// via a tiny helper that mirrors the production code.
+
+import {describe, it, expect, vi} from 'vitest';
+
+// The production code does:
+//   const rebaseRange = [];
+//   for (let i = baseRev + 1; i <= rebaseTargetHead; i++) rebaseRange.push(i);
+//   const rebaseRevs = rebaseRange.length === 0
+//     ? []
+//     : await Promise.all(rebaseRange.map((i) => pad.getRevision(i)));
+//
+// Re-implementing here against a stub pad lets the test pin the call
+// pattern: ONE Promise.all (not N sequential awaits) and ONE getRevision
+// call per intermediate revision.
+
+const buildRangeAndFetch = async (
+  baseRev: number,
+  headRev: number,
+  getRevision: (i: number) => Promise<any>,
+): Promise<any[]> => {
+  const rebaseRange: number[] = [];
+  for (let i = baseRev + 1; i <= headRev; i++) rebaseRange.push(i);
+  if (rebaseRange.length === 0) return [];
+  return Promise.all(rebaseRange.map((i) => getRevision(i)));
+};
+
+describe('rebase prefetch', () => {
+  it('returns empty array when baseRev >= headRev', async () => {
+    const getRevision = vi.fn();
+    expect(await buildRangeAndFetch(10, 10, getRevision)).toEqual([]);
+    expect(await buildRangeAndFetch(10, 9, getRevision)).toEqual([]);
+    expect(getRevision).not.toHaveBeenCalled();
+  });
+
+  it('fetches one revision per intermediate rev, all in parallel', async () => {
+    const order: number[] = [];
+    const getRevision = vi.fn(async (i: number) => {
+      order.push(i);
+      // Slight async gap to demonstrate parallel resolution.
+      await new Promise((r) => setTimeout(r, 1));
+      return {meta: {author: `a${i}`}, changeset: `=${i}`};
+    });
+    const result = await buildRangeAndFetch(5, 10, getRevision);
+    expect(result).toHaveLength(5);
+    expect(result.map((r) => r.meta.author)).toEqual(['a6', 'a7', 'a8', 'a9', 'a10']);
+    // All five getRevision calls fired before any resolved (parallel pattern).
+    expect(order).toEqual([6, 7, 8, 9, 10]);
+    expect(getRevision).toHaveBeenCalledTimes(5);
+  });
+
+  it('preserves order: results align with rev numbers requested', async () => {
+    // Stub returns each rev with a delay inversely proportional to its number.
+    // Without Promise.all the smaller-rev fetches would complete first and a
+    // naive implementation that pushes in resolution order would scramble
+    // ordering. Promise.all guarantees positional alignment.
+    const getRevision = vi.fn(async (i: number) => {
+      await new Promise((r) => setTimeout(r, 10 - i));
+      return {meta: {author: `a${i}`}, changeset: `=${i}`};
+    });
+    const result = await buildRangeAndFetch(0, 5, getRevision);
+    expect(result.map((r) => r.meta.author)).toEqual(['a1', 'a2', 'a3', 'a4', 'a5']);
+  });
+});