Mirrors/etherpad-lite
1
0
Fork 0
mirror of https://github.com/ether/etherpad-lite.git synced 2026-07-18 00:57:55 +00:00
Code Issues Projects Releases Wiki Activity

feat(scaling): NEW_CHANGES_BATCH — pack multi-rev fan-out into one emit

Browse source 
Identified by the #7756 scaling dive (PR #7765) and confirmed by
the engine.io transport investigation in #7767: socket.io's
polling transport batches multiple queued packets into a single
HTTP response, but the WebSocket transport sends one frame per
packet — even when the engine.io socket has several packets
buffered. At 200 concurrent authors that's ~6,600 individual WS
frames/sec/client, starving the apply path of CPU.

This PR addresses the cost at the application layer: when a
recipient is more than one revision behind, the server packs all
queued revisions into a single NEW_CHANGES_BATCH message instead
of emitting NEW_CHANGES once per rev. The wire payload is the same
information, just consolidated.

Feature-flagged:

- settings.newChangesBatch defaults to false. Production behaviour
  is unchanged.
- When enabled, server emits NEW_CHANGES_BATCH iff a recipient has
  >1 rev pending; single-rev fan-outs stay as NEW_CHANGES (no
  framing overhead for the steady-state case).

Clients are forward-compatible: both collab_client.ts (live editor)
and broadcast.ts (timeslider) now accept either message type and
normalise to a list. Newly-built clients work against any server
regardless of the flag; the back-compat hazard is enabling the flag
on a server while old clients are still connected (documented in
the setting's prose).

Tests: src/tests/backend-new/specs/new-changes-batch.test.ts pins
the server's wire-format decision. 4/4 new + 5/5 existing
prom-instruments stay green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
John McLear 2026-05-15 21:52:02 +01:00
parent 79f525b0a7
commit 1a4fa38b2e
8 changed files with 194 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

69
src/node/handler/PadMessageHandler.ts
View file

@ -964,11 +964,29 @@ 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;
// Collect all queued revisions for this socket.
const pending: Array<{
newRev: number;
changeset: string;
apool: unknown;
author: string;
currentTime: number;
timeDelta: number;
}> = [];
while (sessioninfo.rev < pad.getHeadRevisionNumber()) {
const r = sessioninfo.rev + 1;
let revision = revCache[r];
@ -980,30 +998,41 @@ exports.updatePadClients = async (pad: PadType) => {
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;
}
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');
}
}
} catch (err: any) {
messageLogger.error(`Failed to notify user of new revision: ${err.stack || err}`);
}
}));
};