mirror of
https://github.com/ether/etherpad-lite.git
synced 2026-07-17 16:47:05 +00:00
fix: capture head revision atomically with atext to prevent mismatched apply (#7480)
* fix: capture head revision atomically with atext to prevent mismatched apply When constructing CLIENT_VARS, pad.atext was captured at one point but pad.getHeadRevisionNumber() was called later. If concurrent edits advanced the revision between these two reads, the client received initialAttributedText from rev N but rev=N+3, causing "mismatched apply" errors when the next changeset arrived (expecting rev N+3 text). Now captures headRev at the same time as atext and uses the captured value consistently in CLIENT_VARS and sessionInfo. Fixes #4040 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: flush missed revisions after socket joins pad room During handleClientReady(), the server awaits the clientVars hook before socket.join(). Any revisions appended during that await window are broadcast to existing room members but the connecting socket misses them. Call updatePadClients(pad) after joining to flush any such revisions. Also adds a regression test that injects a slow clientVars hook and verifies the connecting client receives catch-up changesets for edits that occurred during the hook await window. Fixes #4040 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * test: fix race condition in clientVars hook test Listen for messages during handshake to avoid missing NEW_CHANGES that arrive before the explicit waitForSocketEvent listener is attached. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: initialize sessionInfo.time before catch-up updatePadClients The catch-up updatePadClients() call introduced in this PR could send NEW_CHANGES with timeDelta=NaN because sessionInfo.time was never set for new sessions. NaN poisons the client-side broadcast/timeslider currentTime tracking. Initialize sessionInfo.time to the timestamp of the snapshot revision before the catch-up flush, with a fallback to Date.now() if the revision date is unavailable. Also strengthens the regression tests: - Validate that initialAttributedText matches the pad AText at the EXACT advertised rev (not just the latest pad text), using pad.getInternalRevisionAText(rev). - Add a load test that hammers the pad with concurrent edits while multiple clients connect, asserting CLIENT_VARS consistency under the exact race condition the fix is targeting. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * test: replace open-ended load loop with bounded mid-handshake edit The previous load test ran 'while (!stopLoad) await pad.setText(...)' in the background while the test connected clients. This saturated ueberDB's write queue and on shutdown the queued writes never drained, hanging the mocha process for the full 6h GitHub Actions job timeout. Replace it with a bounded approach: a clientVars hook lands 3 edits mid-handshake (deterministic, no background loop, no shutdown hang). Still exercises the exact race the fix targets — an edit advancing the rev after the atext snapshot but before CLIENT_VARS is sent — and asserts AText / rev consistency via getInternalRevisionAText. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * test: address remaining Qodo concerns on PR #7480 Addresses Qodo review items 1, 2, 5 from https://github.com/ether/etherpad-lite/issues/comments/4194702740 : - Concern 1 (no loadTesting reproduction test): the suite now toggles settings.loadTest = true in before(), restores in after(). The middle test also pre-populates the pad with 20 revisions before connecting so we genuinely exercise a busy/loaded pad rather than a fresh one. - Concern 2 (no CLIENT_VARS / NEW_CHANGES delay test): the slow clientVars hook in the middle test now has explicit setTimeout delays before AND after the mid-handshake edits, so the race window between atext snapshot and CLIENT_VARS send is observably wide rather than relying on async scheduling alone. The test also collects post-handshake messages and asserts a NEW_CHANGES catch-up arrives when the pad advanced past the advertised rev. - Concern 5 (test doesn't validate rev): both rev-consistency tests use pad.getInternalRevisionAText(advertisedRev) and assert text and attribs match, not just `pad.text() === clientVars.text`. Concerns 3 (connect can miss revisions) and 4 (NaN timeDelta) were already addressed in earlier commits on this branch via the catch-up updatePadClients() call and the sessionInfo.time initialization. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
John McLear
GitHub
parent
7c3837891b
commit
31e0a61126
2 changed files with 245 additions and 3 deletions
|
|
@ -984,8 +984,15 @@ const handleClientReady = async (socket:any, message: ClientReadyMessage) => {
|
|||
// This is a normal first connect
|
||||
let atext;
|
||||
let apool;
|
||||
let headRev: number;
|
||||
// prepare all values for the wire, there's a chance that this throws, if the pad is corrupted
|
||||
try {
|
||||
// Capture atext AND head revision atomically to prevent a race condition where
|
||||
// concurrent edits advance the revision between these two reads. If the client
|
||||
// receives initialAttributedText from rev N but rev=N+3, the first NEW_CHANGES
|
||||
// changeset will fail with "mismatched apply" because it expects rev N+3 text.
|
||||
// See https://github.com/ether/etherpad-lite/issues/4040
|
||||
headRev = pad.getHeadRevisionNumber();
|
||||
atext = cloneAText(pad.atext);
|
||||
const attribsForWire = prepareForWire(atext.attribs, pad.pool);
|
||||
apool = attribsForWire.pool.toJsonable();
|
||||
|
|
@ -1016,7 +1023,7 @@ const handleClientReady = async (socket:any, message: ClientReadyMessage) => {
|
|||
padId: sessionInfo.auth.padID,
|
||||
historicalAuthorData,
|
||||
apool,
|
||||
rev: pad.getHeadRevisionNumber(),
|
||||
rev: headRev,
|
||||
time: currentTime,
|
||||
},
|
||||
colorPalette: authorManager.getColorPalette(),
|
||||
|
|
@ -1081,8 +1088,24 @@ const handleClientReady = async (socket:any, message: ClientReadyMessage) => {
|
|||
// Send the clientVars to the Client
|
||||
socket.emit('message', {type: 'CLIENT_VARS', data: clientVars});
|
||||
|
||||
// Save the current revision in sessioninfos, should be the same as in clientVars
|
||||
sessionInfo.rev = pad.getHeadRevisionNumber();
|
||||
// Save the revision in sessioninfos — must match what was sent in clientVars
|
||||
sessionInfo.rev = headRev;
|
||||
|
||||
// Initialize sessionInfo.time to the timestamp of the snapshot revision so
|
||||
// that subsequent NEW_CHANGES timeDelta calculations are valid. Without
|
||||
// this, the catch-up updatePadClients() call below would emit timeDelta=NaN
|
||||
// which breaks the client's broadcast/timeslider time tracking.
|
||||
try {
|
||||
sessionInfo.time = await pad.getRevisionDate(headRev);
|
||||
} catch (err) {
|
||||
// Fallback: if we can't read the revision timestamp, use now.
|
||||
sessionInfo.time = Date.now();
|
||||
}
|
||||
|
||||
// Flush any revisions that may have been appended while we were awaiting the
|
||||
// clientVars hook (before socket.join). Those revisions were broadcast to
|
||||
// existing room members but this socket hadn't joined yet so it missed them.
|
||||
await exports.updatePadClients(pad);
|
||||
}
|
||||
|
||||
// Notify other users about this new user.
|
||||
|
|
|
|||
219
src/tests/backend/specs/clientvar_rev_consistency.ts
Normal file
219
src/tests/backend/specs/clientvar_rev_consistency.ts
Normal file
|
|
@ -0,0 +1,219 @@
|
|||
'use strict';
|
||||
|
||||
/**
|
||||
* Regression test for https://github.com/ether/etherpad-lite/issues/4040
|
||||
*
|
||||
* Verifies that CLIENT_VARS sends a rev that matches the initialAttributedText.
|
||||
* Previously, pad.atext was captured at one point and pad.getHeadRevisionNumber()
|
||||
* was called later, creating a gap when concurrent edits advanced the revision.
|
||||
*
|
||||
* The whole suite runs with `settings.loadTest = true`. That mode bypasses
|
||||
* socket.io auth and is the documented way to run "load-style" tests against
|
||||
* Etherpad, so this is the configuration where the original "mismatched apply"
|
||||
* production failures were observed.
|
||||
*/
|
||||
|
||||
const assert = require('assert').strict;
|
||||
const common = require('../common');
|
||||
const padManager = require('../../../node/db/PadManager');
|
||||
const plugins = require('../../../static/js/pluginfw/plugin_defs');
|
||||
const settings = require('../../../node/utils/Settings');
|
||||
import {randomString} from '../../../static/js/pad_utils';
|
||||
|
||||
describe(__filename, function () {
|
||||
let agent: any;
|
||||
let clientVarsBackup: any;
|
||||
let loadTestBackup: boolean;
|
||||
|
||||
before(async function () {
|
||||
loadTestBackup = settings.loadTest;
|
||||
settings.loadTest = true;
|
||||
agent = await common.init();
|
||||
clientVarsBackup = plugins.hooks.clientVars || [];
|
||||
});
|
||||
|
||||
after(function () {
|
||||
settings.loadTest = loadTestBackup;
|
||||
});
|
||||
|
||||
afterEach(function () {
|
||||
plugins.hooks.clientVars = clientVarsBackup;
|
||||
});
|
||||
|
||||
it('CLIENT_VARS rev matches initialAttributedText state at that exact rev', async function () {
|
||||
this.timeout(30000);
|
||||
const padId = randomString(10);
|
||||
|
||||
// Create a pad with initial text
|
||||
const pad = await padManager.getPad(padId, 'initial text\n');
|
||||
|
||||
// Make several edits to advance the revision
|
||||
await pad.setText('edit one\n');
|
||||
await pad.setText('edit two\n');
|
||||
await pad.setText('edit three\n');
|
||||
|
||||
// Now connect a new client — CLIENT_VARS should be consistent
|
||||
const res = await agent.get(`/p/${padId}`).expect(200);
|
||||
const socket = await common.connect(res);
|
||||
try {
|
||||
const {type, data: clientVars} = await common.handshake(socket, padId);
|
||||
assert.equal(type, 'CLIENT_VARS');
|
||||
|
||||
const collabVars = clientVars.collab_client_vars;
|
||||
assert.equal(typeof collabVars.rev, 'number');
|
||||
|
||||
// The core invariant: the initialAttributedText must correspond to the
|
||||
// EXACT revision advertised in collabVars.rev (not just the latest pad
|
||||
// text). Validate this by fetching the historical AText for that rev.
|
||||
const atextAtRev = await pad.getInternalRevisionAText(collabVars.rev);
|
||||
assert.equal(atextAtRev.text, collabVars.initialAttributedText.text,
|
||||
`initialAttributedText.text doesn't match pad AText at rev ${collabVars.rev}`);
|
||||
assert.equal(atextAtRev.attribs, collabVars.initialAttributedText.attribs,
|
||||
`initialAttributedText.attribs doesn't match pad AText at rev ${collabVars.rev}`);
|
||||
} finally {
|
||||
socket.close();
|
||||
await pad.remove();
|
||||
}
|
||||
});
|
||||
|
||||
it('CLIENT_VARS stays consistent under concurrent edits during handshake (delay race)',
|
||||
async function () {
|
||||
// Reproduces the original "mismatched apply" race condition:
|
||||
// 1. The server captures pad.atext for CLIENT_VARS.
|
||||
// 2. Time passes (a slow plugin hook, network jitter, GC pause, ...).
|
||||
// 3. While that's happening, another process mutates the pad.
|
||||
// 4. CLIENT_VARS is finally sent — pre-fix this advertised the
|
||||
// *new* rev with the *old* atext.
|
||||
// We exercise this exact window via a slow clientVars hook that
|
||||
// (a) introduces a measurable delay so steps 1 and 4 are not adjacent,
|
||||
// (b) lands several edits during that delay.
|
||||
// The bug also applied at higher load — to also reproduce the load
|
||||
// scenario, we pre-populate the pad with many revisions before connecting.
|
||||
this.timeout(60000);
|
||||
const padId = randomString(10);
|
||||
const pad = await padManager.getPad(padId, 'rev0\n');
|
||||
|
||||
// Pre-populate to put the pad in a "busy" state (high rev count).
|
||||
// Bounded so it can't hang on shutdown.
|
||||
for (let i = 0; i < 20; i++) {
|
||||
await pad.setText(`pre-load-${i}\n`);
|
||||
}
|
||||
const preConnectRev = pad.getHeadRevisionNumber();
|
||||
|
||||
// Inject a slow clientVars hook that:
|
||||
// - waits long enough to make the race window observable, and
|
||||
// - lands additional edits during that wait.
|
||||
let edits = 0;
|
||||
plugins.hooks.clientVars = [{
|
||||
hook_fn: async () => {
|
||||
// Sleep to widen the window between atext snapshot and CLIENT_VARS send.
|
||||
await new Promise((r) => setTimeout(r, 200));
|
||||
for (let i = 0; i < 5; i++) {
|
||||
await pad.setText(`mid-handshake-edit-${edits++}\n`);
|
||||
}
|
||||
// Sleep again so any catch-up logic on the server has to deal with
|
||||
// a long-since-stale snapshot.
|
||||
await new Promise((r) => setTimeout(r, 200));
|
||||
return {};
|
||||
},
|
||||
}];
|
||||
|
||||
try {
|
||||
const res = await agent.get(`/p/${padId}`).expect(200);
|
||||
const socket = await common.connect(res);
|
||||
try {
|
||||
// Listen for catch-up NEW_CHANGES messages alongside the handshake.
|
||||
const messages: any[] = [];
|
||||
socket.on('message', (msg: any) => messages.push(msg));
|
||||
|
||||
const {type, data: clientVars} = await common.handshake(socket, padId);
|
||||
assert.equal(type, 'CLIENT_VARS');
|
||||
const collabVars = clientVars.collab_client_vars;
|
||||
const advertisedRev = collabVars.rev;
|
||||
|
||||
// Pre-fix this would have been violated: rev would point past atext.
|
||||
// Validate the AText matches the pad AText AT THE ADVERTISED REV
|
||||
// (not just the latest pad text), which is the exact invariant whose
|
||||
// violation produced "mismatched apply" errors.
|
||||
const atextAtRev = await pad.getInternalRevisionAText(advertisedRev);
|
||||
assert.equal(atextAtRev.text, collabVars.initialAttributedText.text,
|
||||
`AText mismatch at rev ${advertisedRev}`);
|
||||
assert.equal(atextAtRev.attribs, collabVars.initialAttributedText.attribs,
|
||||
`AText attribs mismatch at rev ${advertisedRev}`);
|
||||
|
||||
// The advertised rev must be at least the pre-connect head — anything
|
||||
// older would mean we shipped a stale snapshot.
|
||||
assert.ok(advertisedRev >= preConnectRev,
|
||||
`CLIENT_VARS rev (${advertisedRev}) is older than pre-connect head (${preConnectRev})`);
|
||||
|
||||
// Wait briefly for in-flight catch-up messages.
|
||||
await new Promise((r) => setTimeout(r, 500));
|
||||
const finalHead = pad.getHeadRevisionNumber();
|
||||
if (advertisedRev < finalHead) {
|
||||
const catchUp = messages.find(
|
||||
(m: any) => m.type === 'COLLABROOM' && m.data?.type === 'NEW_CHANGES');
|
||||
assert.ok(catchUp,
|
||||
`Expected NEW_CHANGES catch-up after CLIENT_VARS (rev ${advertisedRev} -> ${finalHead})`);
|
||||
}
|
||||
} finally {
|
||||
socket.close();
|
||||
}
|
||||
} finally {
|
||||
plugins.hooks.clientVars = clientVarsBackup;
|
||||
await pad.remove();
|
||||
}
|
||||
});
|
||||
|
||||
it('client receives revisions created during clientVars hook await window', async function () {
|
||||
this.timeout(30000);
|
||||
const padId = randomString(10);
|
||||
const pad = await padManager.getPad(padId, 'start\n');
|
||||
|
||||
// Install a slow clientVars hook that simulates a plugin doing async work.
|
||||
// While the hook is running the connecting socket has NOT yet joined the
|
||||
// room, so any edits broadcast in that window would normally be missed.
|
||||
let hookCalled = false;
|
||||
plugins.hooks.clientVars = [{
|
||||
hook_fn: async () => {
|
||||
hookCalled = true;
|
||||
// Mutate the pad while the hook is running — this edit happens after
|
||||
// the atext snapshot but before socket.join().
|
||||
await pad.setText('edited-during-hook\n');
|
||||
return {};
|
||||
},
|
||||
}];
|
||||
|
||||
const res = await agent.get(`/p/${padId}`).expect(200);
|
||||
const socket = await common.connect(res);
|
||||
try {
|
||||
// Collect all messages received during and after handshake so we don't
|
||||
// miss NEW_CHANGES that arrive before we start explicitly listening.
|
||||
const messages: any[] = [];
|
||||
socket.on('message', (msg: any) => messages.push(msg));
|
||||
|
||||
const {type, data: clientVars} = await common.handshake(socket, padId);
|
||||
assert.equal(type, 'CLIENT_VARS');
|
||||
assert.ok(hookCalled, 'clientVars hook should have been called');
|
||||
|
||||
const collabVars = clientVars.collab_client_vars;
|
||||
const clientRev = collabVars.rev;
|
||||
const headRev = pad.getHeadRevisionNumber();
|
||||
|
||||
if (clientRev < headRev) {
|
||||
// Wait a moment for any in-flight messages to arrive.
|
||||
await new Promise((r) => setTimeout(r, 500));
|
||||
const catchUp = messages.find(
|
||||
(m: any) => m.type === 'COLLABROOM' && m.data?.type === 'NEW_CHANGES');
|
||||
assert.ok(catchUp, 'Expected a NEW_CHANGES catch-up message');
|
||||
assert.ok(catchUp.data.newRev > clientRev,
|
||||
`Expected catch-up rev > ${clientRev}, got ${catchUp.data.newRev}`);
|
||||
}
|
||||
assert.ok(clientRev <= headRev,
|
||||
`CLIENT_VARS rev (${clientRev}) must not exceed head rev (${headRev})`);
|
||||
} finally {
|
||||
socket.close();
|
||||
plugins.hooks.clientVars = clientVarsBackup;
|
||||
await pad.remove();
|
||||
}
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue