Skip to content

Commit 9aca0fd

Browse files
Raj85446696Copilot
andauthored
handler: prevent join catch-up revision races (#8409)
* handler: prevent join catch-up revision races Scope the CLIENT_READY catch-up fan-out to the newly joined socket so it cannot advance another session's revision while a queued USER_CHANGES operation is between appendRevision() and its revision assertion. Preserve broadcast behavior for normal edit, API, and import callers. Add a backend regression test that pauses appendRevision(), joins a second client during the pause, and verifies the writer receives ACCEPT_COMMIT without a badChangeset disconnect. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> * test: synchronize join catch-up race regression Wait for the actual updatePadClients() call triggered by the joining handler instead of sleeping after CLIENT_VARS. CLIENT_VARS is emitted before catch-up, so the fixed test now deterministically holds the writer until the raced catch-up has completed and would fail against the original all-socket implementation. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
1 parent a15535b commit 9aca0fd

2 files changed

Lines changed: 103 additions & 3 deletions

File tree

‎src/node/handler/PadMessageHandler.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1039,9 +1039,10 @@ const handleUserChanges = async (socket:any, message: {
10391039
}
10401040
};
10411041

1042-
exports.updatePadClients = async (pad: PadType) => {
1042+
exports.updatePadClients = async (pad: PadType, onlySocketId?: string) => {
10431043
// skip this if no-one is on this pad
1044-
const roomSockets = _getRoomSockets(pad.id);
1044+
let roomSockets = _getRoomSockets(pad.id);
1045+
if (onlySocketId != null) roomSockets = roomSockets.filter((socket) => socket.id === onlySocketId);
10451046
if (roomSockets.length === 0) return;
10461047

10471048
// since all clients usually get the same set of changesets, store them in local cache
@@ -1503,7 +1504,7 @@ const handleClientReady = async (socket:any, message: ClientReadyMessage) => {
15031504
// Flush any revisions that may have been appended while we were awaiting the
15041505
// clientVars hook (before socket.join). Those revisions were broadcast to
15051506
// existing room members but this socket hadn't joined yet so it missed them.
1506-
await exports.updatePadClients(pad);
1507+
await exports.updatePadClients(pad, socket.id);
15071508
}
15081509

15091510
// Notify other users about this new user.
Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
'use strict';
2+
3+
/**
4+
* The catch-up updatePadClients() call at the end of handleClientReady() (added in #7480) runs
5+
* outside the per-pad USER_CHANGES queue. If a client joins while another client's USER_CHANGES
6+
* is between pad.appendRevision() and the `assert.equal(thisSession.rev, r)` check, the catch-up
7+
* advances the writer's session rev and the writer is disconnected with badChangeset.
8+
*/
9+
10+
const assert = require('assert').strict;
11+
const common = require('../common');
12+
const padManager = require('../../../node/db/PadManager');
13+
const padMessageHandler = require('../../../node/handler/PadMessageHandler');
14+
15+
describe(__filename, function () {
16+
let agent: any;
17+
let pad: any;
18+
let padId: string;
19+
const sockets: any[] = [];
20+
let updatePadClientsBackup: any;
21+
22+
before(async function () {
23+
agent = await common.init();
24+
});
25+
26+
beforeEach(async function () {
27+
updatePadClientsBackup = padMessageHandler.updatePadClients;
28+
padId = common.randomString();
29+
pad = await padManager.getPad(padId, '\n');
30+
await pad.setText('\n');
31+
});
32+
33+
afterEach(async function () {
34+
padMessageHandler.updatePadClients = updatePadClientsBackup;
35+
for (const s of sockets.splice(0)) s.close();
36+
if (pad != null) await pad.remove();
37+
pad = null;
38+
});
39+
40+
const join = async () => {
41+
const res = await agent.get(`/p/${padId}`).expect(200);
42+
const socket = await common.connect(res);
43+
sockets.push(socket);
44+
const {type, data} = await common.handshake(socket, padId);
45+
assert.equal(type, 'CLIENT_VARS');
46+
return {socket, rev: data.collab_client_vars.rev, authorId: data.userId};
47+
};
48+
49+
it('a join during another user\'s USER_CHANGES does not reject that change', async function () {
50+
this.timeout(20000);
51+
const writer = await join();
52+
53+
// Hold the writer's USER_CHANGES right after the revision was appended.
54+
let reached: () => void;
55+
const reachedP = new Promise<void>((r) => { reached = r; });
56+
let release: () => void;
57+
const releaseP = new Promise<void>((r) => { release = r; });
58+
const origAppend = pad.appendRevision.bind(pad);
59+
pad.appendRevision = async (...args: any[]) => {
60+
const newRev = await origAppend(...args);
61+
pad.appendRevision = origAppend;
62+
reached();
63+
await releaseP;
64+
return newRev;
65+
};
66+
67+
const writerMsgs: any[] = [];
68+
writer.socket.on('message', (msg: any) => writerMsgs.push(msg));
69+
common.sendUserChanges(writer.socket, {
70+
baseRev: writer.rev,
71+
changeset: 'Z:1>5*0+5$hello',
72+
apool: {numToAttrib: {0: ['author', writer.authorId]}, nextNum: 1},
73+
}).catch(() => {});
74+
await reachedP;
75+
76+
// Somebody joins while the change is in flight.
77+
let catchUpDone: () => void;
78+
const catchUpDoneP = new Promise<void>((r) => { catchUpDone = r; });
79+
padMessageHandler.updatePadClients = async (...args: any[]) => {
80+
const result = await updatePadClientsBackup(...args);
81+
catchUpDone();
82+
return result;
83+
};
84+
await join();
85+
// CLIENT_VARS is emitted before handleClientReady() performs its catch-up.
86+
// Wait for updatePadClients() itself so this test cannot release the writer
87+
// before the join handler reaches the raced operation.
88+
await catchUpDoneP;
89+
release!();
90+
await new Promise((r) => setTimeout(r, 500));
91+
92+
const types = writerMsgs.map((m) => m.disconnect ? `disconnect:${m.disconnect}` : m.data?.type);
93+
// Plugins (e.g. ep_set_title_on_pad) may broadcast CUSTOM messages on join.
94+
const ignored = new Set(['USER_NEWINFO', 'CUSTOM']);
95+
assert.deepEqual(types.filter((t) => !ignored.has(t)), ['ACCEPT_COMMIT'],
96+
`writer got ${JSON.stringify(types)}`);
97+
assert.equal(pad.text(), 'hello\n');
98+
});
99+
});

0 commit comments

Comments
 (0)