diff --git a/src/node/handler/PadMessageHandler.ts b/src/node/handler/PadMessageHandler.ts index 764ec2e6e68..125a7580320 100644 --- a/src/node/handler/PadMessageHandler.ts +++ b/src/node/handler/PadMessageHandler.ts @@ -1056,40 +1056,64 @@ exports.updatePadClients = async (pad: PadType) => { // The user might have disconnected since _getRoomSockets() was called. if (sessioninfo == null) return; - 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; + // One fan-out per socket at a time. Without this, two concurrent + // updatePadClients() runs both read `sessioninfo.rev`, both await + // pad.getRevision(), and both emit the same revision to the same client + // (#7756 lever 3): the loop's read-await-write is not atomic. + // + // The claim is a separate flag on purpose. `sessioninfo.rev` has to keep + // meaning "last revision actually put on the wire" — handleUserChanges + // asserts `thisSession.rev === r` before emitting ACCEPT_COMMIT precisely to + // guarantee the client sees NEW_CHANGES and ACCEPT_COMMIT in order. Bumping + // rev up front to claim a range would satisfy that assert while the + // NEW_CHANGES messages are still queued behind an await, which is the + // out-of-order delivery the assert exists to prevent. + // + // A skipped run loses nothing: the run that holds the flag re-reads + // getHeadRevisionNumber() every iteration, so it also ships whatever arrived + // in the meantime. + if (sessioninfo.fanOutInFlight) return; + sessioninfo.fanOutInFlight = true; + try { + 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 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; + 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; + } + sessioninfo.time = currentTime; + sessioninfo.rev = r; } - sessioninfo.time = currentTime; - sessioninfo.rev = r; + } finally { + // Cleared even on the `return` inside the emit-failure path above, so a + // failed send doesn't wedge the socket's fan-out for good. + sessioninfo.fanOutInFlight = false; } })); }; diff --git a/src/tests/backend/specs/padFanOutSerialization.ts b/src/tests/backend/specs/padFanOutSerialization.ts new file mode 100644 index 00000000000..bdca46699c8 --- /dev/null +++ b/src/tests/backend/specs/padFanOutSerialization.ts @@ -0,0 +1,101 @@ +'use strict'; + +/** + * updatePadClients() fan-out must not double-send a revision to a client + * (#7756 lever 3). + * + * The per-socket loop reads `sessioninfo.rev`, awaits `pad.getRevision()`, then + * writes `sessioninfo.rev` back. That read-await-write is not atomic, so two + * concurrent fan-outs for the same pad both start from the same rev and both emit + * it. The client applies the changeset twice and its revision numbering diverges + * from the server's. + * + * Guarded by `sessioninfo.fanOutInFlight`, deliberately NOT by pre-claiming + * `sessioninfo.rev`: handleUserChanges asserts `thisSession.rev === r` to keep + * NEW_CHANGES and ACCEPT_COMMIT ordered on the wire, and that assert is only + * meaningful while `rev` means "already sent". + */ + +import {PadType} from '../../../node/types/PadType'; + +const assert = require('assert').strict; +const common = require('../common'); +const padManager = require('../../../node/db/PadManager'); +const padMessageHandler = require('../../../node/handler/PadMessageHandler'); + +let agent: any; + +describe(__filename, function () { + this.timeout(30000); + + let pad: PadType; + let padId: string; + let socket: any; + + before(async function () { + agent = await common.init(); + }); + + beforeEach(async function () { + padId = common.randomString(); + pad = await padManager.getPad(padId, 'x\n'); + const res = await agent.get(`/p/${padId}`).expect(200); + socket = await common.connect(res); + const {type} = await common.handshake(socket, padId); + assert.equal(type, 'CLIENT_VARS'); + }); + + afterEach(async function () { + if (socket != null) socket.close(); + socket = null; + await padManager.getPad(padId).then((p: PadType) => p.remove()); + }); + + const collectNewChanges = (ms: number) => { + const revs: number[] = []; + const onMessage = (msg: any) => { + if (msg?.type === 'COLLABROOM' && msg?.data?.type === 'NEW_CHANGES') { + revs.push(msg.data.newRev); + } + }; + socket.on('message', onMessage); + return new Promise((resolve) => setTimeout(() => { + socket.off('message', onMessage); + resolve(revs); + }, ms)); + }; + + it('sends each revision exactly once when fan-outs overlap', async function () { + const collected = collectNewChanges(2000); + // Append behind the client's back so the pad has a backlog to fan out, then + // race two fan-outs. Without the guard both runs start at the same rev. + for (const text of ['a\n', 'b\n', 'c\n']) await pad.setText(text); + const head = pad.getHeadRevisionNumber(); + await Promise.all([ + padMessageHandler.updatePadClients(pad), + padMessageHandler.updatePadClients(pad), + padMessageHandler.updatePadClients(pad), + ]); + + const revs = await collected; + assert.deepEqual([...new Set(revs)], revs, + `every revision must be sent once, got ${JSON.stringify(revs)}`); + assert.deepEqual(revs, [...revs].sort((a, b) => a - b), + `revisions must arrive in order, got ${JSON.stringify(revs)}`); + assert.equal(revs[revs.length - 1], head, 'the client is caught up to head'); + }); + + it('releases the in-flight flag so later fan-outs still deliver', async function () { + // Collect across both fan-outs: a flag that is never cleared would deliver + // the first revision and then go silent. + const collected = collectNewChanges(2000); + await pad.setText('first\n'); + await padMessageHandler.updatePadClients(pad); + await pad.setText('second\n'); + const head = pad.getHeadRevisionNumber(); + await padMessageHandler.updatePadClients(pad); + const revs = await collected; + assert.deepEqual(revs, [head - 1, head], + `both revisions must be delivered, got ${JSON.stringify(revs)}`); + }); +});