From 3565f51a84134ccab55acee73758138e20bd8b39 Mon Sep 17 00:00:00 2001 From: Jakob Ackermann Date: Tue, 21 Jul 2026 11:11:21 +0200 Subject: [PATCH] [monorepo] prepare for per-project pub/sub for applied ops (#35663) GitOrigin-RevId: 11d45addf366c3449cad432272ed03d4c8dae49d --- .../app/js/HistoryOTUpdateManager.js | 2 +- .../app/js/RealTimeRedisManager.js | 32 ++- .../document-updater/app/js/UpdateManager.js | 2 +- .../config/settings.defaults.js | 10 + .../js/ApplyingUpdatesToADocTests.js | 79 ++++++++ .../acceptance/js/helpers/DocUpdaterClient.js | 10 +- .../RealTimeRedisManagerTests.js | 122 ++++++++++-- .../app/js/DocumentUpdaterController.js | 81 +++++--- .../real-time/app/js/WebsocketLoadBalancer.js | 23 +++ .../test/acceptance/js/MatrixTests.js | 28 +++ .../test/acceptance/js/ReceiveUpdateTests.js | 129 +++++++++++- .../unit/js/DocumentUpdaterController.test.js | 183 ++++++++++++++++-- .../unit/js/WebsocketLoadBalancer.test.js | 93 ++++++++- 13 files changed, 723 insertions(+), 71 deletions(-) diff --git a/services/document-updater/app/js/HistoryOTUpdateManager.js b/services/document-updater/app/js/HistoryOTUpdateManager.js index a3298b592c..76ccc2bf90 100644 --- a/services/document-updater/app/js/HistoryOTUpdateManager.js +++ b/services/document-updater/app/js/HistoryOTUpdateManager.js @@ -151,7 +151,7 @@ async function applyUpdate(projectId, docId, update) { RealTimeRedisManager.sendData({ project_id: projectId, doc_id: docId, - error: error instanceof Error ? error.message : error, + error: error instanceof Error ? error.message : String(error), }) profiler.log('sendData') throw error diff --git a/services/document-updater/app/js/RealTimeRedisManager.js b/services/document-updater/app/js/RealTimeRedisManager.js index 1fac34c779..5badc79246 100644 --- a/services/document-updater/app/js/RealTimeRedisManager.js +++ b/services/document-updater/app/js/RealTimeRedisManager.js @@ -140,22 +140,46 @@ const RealTimeRedisManager = { }, /** - * Publish an applied-ops message, stamped with a unique message id. + * Publish an applied op or an error for a doc back to real-time. * * @param {Object} data + * @param {string} data.project_id - routes the message to the project's + * editor-events channel (when publishAppliedOpsOnEditorEvents is + * enabled) and to the project room in real-time + * @param {string} data.doc_id - routes the message to the doc's applied-ops + * channel (legacy path) and identifies the doc for clients + * @param {Object} [data.op] - the applied op + * @param {string} [data.error] - the error message when applying failed */ sendData(data) { // create a unique message id using a counter const messageId = `doc:${HOST}:${RND}-${COUNT++}` - if (data != null) { - data._id = messageId + data._id = messageId + + // Send applied ops on the per-project editor-events channel, which + // real-time subscribes to for the lifetime of every connected project, + // instead of the legacy per-doc applied-ops channel. + const path = Settings.publishAppliedOpsOnEditorEvents + ? 'editor-events' + : 'applied-ops' + if (path === 'editor-events') { + // the message name routes the message to the applied-ops handling in + // real-time's editor-events processing + data.message = data.op ? 'otUpdateApplied' : 'otUpdateError' } const blob = JSON.stringify(data) - metrics.summary('redis.publish.applied-ops', blob.length) + metrics.summary('redis.publish.applied-ops', blob.length, { path }) // publish on separate channels for individual projects and docs when // configured (needs realtime to be configured for this too). + if (path === 'editor-events') { + if (Settings.publishOnIndividualChannels) { + return pubsubClient.publish(`editor-events:${data.project_id}`, blob) + } else { + return pubsubClient.publish('editor-events', blob) + } + } if (Settings.publishOnIndividualChannels) { return pubsubClient.publish(`applied-ops:${data.doc_id}`, blob) } else { diff --git a/services/document-updater/app/js/UpdateManager.js b/services/document-updater/app/js/UpdateManager.js index e8cf51da6e..1eac2d47e2 100644 --- a/services/document-updater/app/js/UpdateManager.js +++ b/services/document-updater/app/js/UpdateManager.js @@ -266,7 +266,7 @@ const UpdateManager = { RealTimeRedisManager.sendData({ project_id: projectId, doc_id: docId, - error: error instanceof Error ? error.message : error, + error: error instanceof Error ? error.message : String(error), }) profile.log('sendData') throw error diff --git a/services/document-updater/config/settings.defaults.js b/services/document-updater/config/settings.defaults.js index 6645c72c8a..2363de7255 100755 --- a/services/document-updater/config/settings.defaults.js +++ b/services/document-updater/config/settings.defaults.js @@ -185,6 +185,16 @@ module.exports = { publishOnIndividualChannels: process.env.PUBLISH_ON_INDIVIDUAL_CHANNELS === 'true', + // Migration of the applied-ops pub/sub (document-updater -> real-time) from + // the per-doc applied-ops channel to the per-project editor-events channel. + // See docs/per-project-applied-ops-pubsub-migration.md. Requires all + // real-time instances to handle applied-ops on editor-events before being + // enabled. Composes with publishOnIndividualChannels, which picks between + // the base channel and the per-doc/per-project channel of the family + // selected here. + publishAppliedOpsOnEditorEvents: + process.env.PUBLISH_APPLIED_OPS_ON_EDITOR_EVENTS === 'true', + continuousBackgroundFlush: process.env.CONTINUOUS_BACKGROUND_FLUSH === 'true', smoothingOffset: process.env.SMOOTHING_OFFSET || 1000, // milliseconds diff --git a/services/document-updater/test/acceptance/js/ApplyingUpdatesToADocTests.js b/services/document-updater/test/acceptance/js/ApplyingUpdatesToADocTests.js index 0bac16df4b..a502c4a9e3 100644 --- a/services/document-updater/test/acceptance/js/ApplyingUpdatesToADocTests.js +++ b/services/document-updater/test/acceptance/js/ApplyingUpdatesToADocTests.js @@ -658,6 +658,85 @@ describe('Applying updates to a doc', function () { }) }) + describe('with publishAppliedOpsOnEditorEvents enabled', function () { + beforeEach(function () { + Settings.publishAppliedOpsOnEditorEvents = true + }) + + afterEach(function () { + Settings.publishAppliedOpsOnEditorEvents = false + }) + + describe('with an applied update', function () { + beforeEach(async function () { + MockWebApi.insertDoc(this.project_id, this.doc_id, { + lines: this.lines, + version: this.version, + }) + + DocUpdaterClient.subscribeToAppliedOps( + (this.messageCallback = sinon.stub()) + ) + + await sendUpdateAndWait(this.project_id, this.doc_id, this.update) + }) + + it('should update the doc', async function () { + const doc = await DocUpdaterClient.getDoc(this.project_id, this.doc_id) + doc.lines.should.deep.equal(this.result) + }) + + it('should publish the applied op on the editor-events channel', function () { + this.messageCallback.called.should.equal(true) + const [channel, message] = this.messageCallback.args[0] + channel.should.equal('editor-events') + const parsedMessage = JSON.parse(message) + parsedMessage.should.deep.include({ + project_id: this.project_id, + doc_id: this.doc_id, + message: 'otUpdateApplied', + }) + parsedMessage.op.v.should.equal(this.version) + }) + }) + + describe('with a broken update', function () { + beforeEach(async function () { + this.broken_update = { + doc: this.doc_id, + v: this.version, + op: [{ d: 'not the correct content', p: 0 }], + } + MockWebApi.insertDoc(this.project_id, this.doc_id, { + lines: this.lines, + version: this.version, + }) + + DocUpdaterClient.subscribeToAppliedOps( + (this.messageCallback = sinon.stub()) + ) + + await sendUpdateAndWait( + this.project_id, + this.doc_id, + this.broken_update + ) + }) + + it('should publish the error on the editor-events channel', function () { + this.messageCallback.called.should.equal(true) + const [channel, message] = this.messageCallback.args[0] + channel.should.equal('editor-events') + JSON.parse(message).should.deep.include({ + project_id: this.project_id, + doc_id: this.doc_id, + message: 'otUpdateError', + error: 'Delete component does not match', + }) + }) + }) + }) + describe('with a broken update (history-ot)', function () { beforeEach(async function () { this.broken_update = { diff --git a/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js b/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js index 4646c11022..bf68dfd80b 100644 --- a/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js +++ b/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js @@ -12,6 +12,9 @@ const rclientSub = require('@overleaf/redis-wrapper').createClient( Settings.redis.pubsub ) rclientSub.subscribe('applied-ops') +// applied ops are published on the editor-events channel instead when +// publishAppliedOpsOnEditorEvents is enabled +rclientSub.subscribe('editor-events') rclientSub.setMaxListeners(0) function getPendingUpdateListKey() { @@ -33,7 +36,12 @@ module.exports = DocUpdaterClient = { }, subscribeToAppliedOps(messageHandler) { - rclientSub.on('message', messageHandler) + rclientSub.on('message', (channel, message) => { + // every applied op also publishes a bandwidth-estimation canary on the + // editor-events channel; drop those so tests only see the applied ops + if (message.includes('"canary-applied-op"')) return + messageHandler(channel, message) + }) }, // Enqueue an update for the doc-updater: push the payload (tagged with its diff --git a/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js b/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js index 99c3b7fe1a..90544207d6 100644 --- a/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js +++ b/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js @@ -28,7 +28,7 @@ describe('RealTimeRedisManager', function () { createClient: config => config.name === 'pubsub' ? this.pubsubClient : this.rclient, }, - '@overleaf/settings': { + '@overleaf/settings': (this.Settings = { redis: { documentupdater: (this.settings = { key_schema: { @@ -41,7 +41,7 @@ describe('RealTimeRedisManager', function () { name: 'pubsub', }, }, - }, + }), crypto: (this.crypto = { randomBytes: sinon .stub() @@ -143,25 +143,113 @@ describe('RealTimeRedisManager', function () { return describe('sendData', function () { beforeEach(function () { this.message_id = 'doc:somehost:01020304-0' - return this.RealTimeRedisManager.sendData({ op: 'thisop' }) + this.data = { + project_id: this.project_id, + doc_id: this.doc_id, + op: 'thisop', + } + this.blob = JSON.stringify({ ...this.data, _id: this.message_id }) }) - it('should send the op with a message id', function () { - return this.pubsubClient.publish - .calledWith( - 'applied-ops', - JSON.stringify({ op: 'thisop', _id: this.message_id }) - ) - .should.equal(true) + describe('on the base applied-ops channel', function () { + beforeEach(function () { + this.RealTimeRedisManager.sendData(this.data) + }) + + it('should send the op with a message id', function () { + return this.pubsubClient.publish + .calledWith('applied-ops', this.blob) + .should.equal(true) + }) + + it('should track the payload size', function () { + return this.metrics.summary + .calledWith('redis.publish.applied-ops', this.blob.length, { + path: 'applied-ops', + }) + .should.equal(true) + }) }) - return it('should track the payload size', function () { - return this.metrics.summary - .calledWith( - 'redis.publish.applied-ops', - JSON.stringify({ op: 'thisop', _id: this.message_id }).length - ) - .should.equal(true) + describe('on the per-doc applied-ops channel', function () { + beforeEach(function () { + this.Settings.publishOnIndividualChannels = true + this.RealTimeRedisManager.sendData(this.data) + }) + + it('should send the op on the per-doc channel', function () { + return this.pubsubClient.publish + .calledWith(`applied-ops:${this.doc_id}`, this.blob) + .should.equal(true) + }) + }) + + describe('on the base editor-events channel', function () { + beforeEach(function () { + this.Settings.publishAppliedOpsOnEditorEvents = true + this.blob = JSON.stringify({ + ...this.data, + _id: this.message_id, + message: 'otUpdateApplied', + }) + this.RealTimeRedisManager.sendData(this.data) + }) + + it('should send the op with a message id and message name', function () { + return this.pubsubClient.publish + .calledWith('editor-events', this.blob) + .should.equal(true) + }) + + it('should track the payload size', function () { + return this.metrics.summary + .calledWith('redis.publish.applied-ops', this.blob.length, { + path: 'editor-events', + }) + .should.equal(true) + }) + }) + + describe('on the per-project editor-events channel', function () { + beforeEach(function () { + this.Settings.publishAppliedOpsOnEditorEvents = true + this.Settings.publishOnIndividualChannels = true + this.blob = JSON.stringify({ + ...this.data, + _id: this.message_id, + message: 'otUpdateApplied', + }) + this.RealTimeRedisManager.sendData(this.data) + }) + + it('should send the op on the per-project channel', function () { + return this.pubsubClient.publish + .calledWith(`editor-events:${this.project_id}`, this.blob) + .should.equal(true) + }) + }) + + describe('with an error on the editor-events channel', function () { + beforeEach(function () { + this.Settings.publishAppliedOpsOnEditorEvents = true + this.data = { + project_id: this.project_id, + doc_id: this.doc_id, + error: 'something went wrong', + } + this.blob = JSON.stringify({ + ...this.data, + _id: this.message_id, + message: 'otUpdateError', + }) + this.RealTimeRedisManager.sendData(this.data) + }) + + it('should send the error with the otUpdateError message name', function () { + return this.pubsubClient.publish + .calledWith('editor-events', this.blob) + .should.equal(true) + }) }) }) }) diff --git a/services/real-time/app/js/DocumentUpdaterController.js b/services/real-time/app/js/DocumentUpdaterController.js index 397872479e..7cf91a3659 100644 --- a/services/real-time/app/js/DocumentUpdaterController.js +++ b/services/real-time/app/js/DocumentUpdaterController.js @@ -66,42 +66,61 @@ export default DocumentUpdaterController = { logger.error({ err: error, channel }, 'error parsing JSON') return } - if (message.op) { - if (message._id && settings.checkEventOrder) { - const status = EventLogger.checkEventOrder( - 'applied-ops', - message._id, - message - ) - if (status === 'duplicate') { - return // skip duplicate events - } - } - DocumentUpdaterController._applyUpdateFromDocumentUpdater( - io, - message.doc_id, - message.op - ) - } else if (message.error) { - DocumentUpdaterController._processErrorFromDocumentUpdater( - io, - message.doc_id, - message.error, - message - ) - } else if (message.health_check) { + if (message.health_check) { logger.debug( { message }, 'got health check message in applied ops channel' ) HealthCheckManager.check(channel, message.key) + return } + // Messages on the legacy applied-ops channel are broadcast to the doc + // room, matching the per-doc channel granularity. + DocumentUpdaterController.handleAppliedOpMessage( + io, + message, + message.doc_id + ) }) }, - _applyUpdateFromDocumentUpdater(io, docId, update) { + // Handle an already-parsed message from document-updater (an applied op + // or an error) and broadcast it to the given room: the doc room for + // messages from the legacy per-doc applied-ops channel, the project room + // for messages forwarded from the editor-events channel by the + // WebsocketLoadBalancer. + handleAppliedOpMessage(io, message, roomId) { + if (message.op) { + if (message._id && settings.checkEventOrder) { + const status = EventLogger.checkEventOrder( + 'applied-ops', + message._id, + message + ) + if (status === 'duplicate') { + return // skip duplicate events + } + } + DocumentUpdaterController._applyUpdateFromDocumentUpdater( + io, + roomId, + message.doc_id, + message.op + ) + } else if (message.error) { + DocumentUpdaterController._processErrorFromDocumentUpdater( + io, + roomId, + message.doc_id, + message.error, + message + ) + } + }, + + _applyUpdateFromDocumentUpdater(io, roomId, docId, update) { let client - const clientList = io.sockets.clients(docId) + const clientList = io.sockets.clients(roomId) // avoid unnecessary work if no clients are connected if (clientList.length === 0) { return @@ -172,8 +191,14 @@ export default DocumentUpdaterController = { } }, - _processErrorFromDocumentUpdater(io, docId, error, message) { - for (const client of io.sockets.clients(docId)) { + _processErrorFromDocumentUpdater(io, roomId, docId, error, message) { + for (const client of io.sockets.clients(roomId)) { + // When broadcasting via the project room, only clients that have + // joined the doc are affected by the error. (Always true for clients + // in the doc room.) + if (client.ol_context?.[`doc:${docId}`] !== 'allowed') { + continue + } logger.warn( { err: error, docId, clientId: client.id }, 'error from document updater, disconnecting client' diff --git a/services/real-time/app/js/WebsocketLoadBalancer.js b/services/real-time/app/js/WebsocketLoadBalancer.js index f6a7d78c89..a6c4549c35 100644 --- a/services/real-time/app/js/WebsocketLoadBalancer.js +++ b/services/real-time/app/js/WebsocketLoadBalancer.js @@ -8,6 +8,7 @@ import HealthCheckManager from './HealthCheckManager.js' import RoomManager from './RoomManager.js' import ChannelManager from './ChannelManager.js' import ConnectedUsersManager from './ConnectedUsersManager.js' +import DocumentUpdaterController from './DocumentUpdaterController.js' const RESTRICTED_USER_MESSAGE_TYPE_PASS_LIST = [ 'otUpdateApplied', @@ -171,6 +172,28 @@ export default WebsocketLoadBalancer = { estimateBandwidth(projectId, 'per-project') estimateBandwidth(docId, 'per-doc') + } else if ( + message.message === 'otUpdateApplied' || + message.message === 'otUpdateError' + ) { + // An applied op (or error) from document-updater, published on the + // per-project editor-events channel instead of the legacy per-doc + // applied-ops channel. Forward it for broadcasting to the project + // room. + const status = message.message === 'otUpdateApplied' ? 'op' : 'error' + Metrics.inc('applied-ops-via-editor-events', 1, { status }) + if (!message.project_id) { + logger.error( + { channel, message }, + 'applied-ops message without project_id' + ) + } else { + DocumentUpdaterController.handleAppliedOpMessage( + io, + message, + message.project_id + ) + } } else if (message.room_id) { if (message._id && Settings.checkEventOrder) { const status = EventLogger.checkEventOrder( diff --git a/services/real-time/test/acceptance/js/MatrixTests.js b/services/real-time/test/acceptance/js/MatrixTests.js index 69ae06950e..1ee86c21d7 100644 --- a/services/real-time/test/acceptance/js/MatrixTests.js +++ b/services/real-time/test/acceptance/js/MatrixTests.js @@ -389,6 +389,34 @@ describe('MatrixTests', function () { }) }) + describe('receive updates via editor-events', function () { + const receivedMessages = [] + beforeEach(function publishAnUpdateInRedis(done) { + const update = { + project_id: privateProjectId, + doc_id: privateDocId, + message: 'otUpdateApplied', + op: { + meta: { source: privateClient.publicId }, + v: 42, + doc: privateDocId, + op: [{ i: 'foo', p: 50 }], + }, + } + client.on('otUpdateApplied', update => { + receivedMessages.push(update) + }) + privateClient.once('otUpdateApplied', () => { + setTimeout(done, 10) + }) + rclient.publish('editor-events', JSON.stringify(update)) + }) + + it('should send nothing to client', function () { + expect(receivedMessages).to.have.length(0) + }) + }) + describe('receive messages from web', function () { const receivedMessages = [] beforeEach(function publishAMessageInRedis(done) { diff --git a/services/real-time/test/acceptance/js/ReceiveUpdateTests.js b/services/real-time/test/acceptance/js/ReceiveUpdateTests.js index a7b07835c4..2edf5d9219 100644 --- a/services/real-time/test/acceptance/js/ReceiveUpdateTests.js +++ b/services/real-time/test/acceptance/js/ReceiveUpdateTests.js @@ -69,6 +69,11 @@ describe('receiveUpdate', function () { return this.clientB.emit('joinDoc', this.doc_id, cb) }, + cb => { + // clientD is in the first project, but has not joined the doc + this.clientD = RealTimeClient.connect(this.project_id, cb) + }, + cb => { return FixturesManager.setUpProject( { @@ -116,6 +121,10 @@ describe('receiveUpdate', function () { this.clientC.on('otUpdateApplied', update => this.clientCUpdates.push(update) ) + this.clientDUpdates = [] + this.clientD.on('otUpdateApplied', update => + this.clientDUpdates.push(update) + ) this.clientAErrors = [] this.clientA.on('otUpdateError', error => @@ -129,6 +138,10 @@ describe('receiveUpdate', function () { this.clientC.on('otUpdateError', error => this.clientCErrors.push(error) ) + this.clientDErrors = [] + this.clientD.on('otUpdateError', error => + this.clientDErrors.push(error) + ) return cb() }, ], @@ -143,7 +156,10 @@ describe('receiveUpdate', function () { if (this.clientB != null) { this.clientB.disconnect() } - return this.clientC != null ? this.clientC.disconnect() : undefined + if (this.clientC != null) { + this.clientC.disconnect() + } + return this.clientD != null ? this.clientD.disconnect() : undefined }) describe('with an update from clientA', function () { @@ -246,6 +262,117 @@ describe('receiveUpdate', function () { }) }) + describe('with an update from clientA published on the editor-events channel', function () { + beforeEach(function (done) { + this.update = { + project_id: this.project_id, + doc_id: this.doc_id, + message: 'otUpdateApplied', + op: { + meta: { + source: this.clientA.publicId, + }, + v: this.version, + doc: this.doc_id, + op: [{ i: 'foo', p: 50 }], + }, + } + rclient.publish( + `editor-events:${this.project_id}`, + JSON.stringify(this.update) + ) + setTimeout(done, 200) + }) // Give clients time to get message + + it('should send the full op to clientB', function () { + this.clientBUpdates.should.deep.equal([this.update.op]) + }) + + it('should send an ack to clientA', function () { + this.clientAUpdates.should.deep.equal([ + { + v: this.version, + doc: this.doc_id, + }, + ]) + }) + + it('should send the full op to clientD, who has not joined the doc', function () { + this.clientDUpdates.should.deep.equal([this.update.op]) + }) + + it('should send nothing to clientC', function () { + this.clientCUpdates.should.deep.equal([]) + }) + }) + + describe('with an update from a remote client published on the base editor-events channel', function () { + // producers with PUBLISH_ON_INDIVIDUAL_CHANNELS disabled publish on the + // base channel instead of the per-project channel + beforeEach(function (done) { + this.update = { + project_id: this.project_id, + doc_id: this.doc_id, + message: 'otUpdateApplied', + op: { + meta: { + source: 'this-is-a-remote-client-id', + }, + v: this.version, + doc: this.doc_id, + op: [{ i: 'foo', p: 50 }], + }, + } + rclient.publish('editor-events', JSON.stringify(this.update)) + setTimeout(done, 200) + }) // Give clients time to get message + + it('should send the full op to the clients in the first project', function () { + this.clientAUpdates.should.deep.equal([this.update.op]) + this.clientBUpdates.should.deep.equal([this.update.op]) + this.clientDUpdates.should.deep.equal([this.update.op]) + }) + + it('should send nothing to clientC', function () { + this.clientCUpdates.should.deep.equal([]) + }) + }) + + describe('with an error for the first project published on the editor-events channel', function () { + beforeEach(function (done) { + rclient.publish( + `editor-events:${this.project_id}`, + JSON.stringify({ + project_id: this.project_id, + doc_id: this.doc_id, + message: 'otUpdateError', + error: (this.error = 'something went wrong'), + }) + ) + setTimeout(done, 200) + }) // Give clients time to get message + + it('should send the error to the clients that joined the doc', function () { + this.clientAErrors.should.deep.equal([this.error]) + this.clientBErrors.should.deep.equal([this.error]) + }) + + it('should disconnect the clients that joined the doc', function () { + this.clientA.socket.connected.should.equal(false) + this.clientB.socket.connected.should.equal(false) + }) + + it('should not send the error to clientD, who has not joined the doc', function () { + this.clientDErrors.should.deep.equal([]) + this.clientD.socket.connected.should.equal(true) + }) + + it('should not send any errors to the client in the second project', function () { + this.clientCErrors.should.deep.equal([]) + this.clientC.socket.connected.should.equal(true) + }) + }) + describe('with an error for the first project', function () { beforeEach(function (done) { rclient.publish( diff --git a/services/real-time/test/unit/js/DocumentUpdaterController.test.js b/services/real-time/test/unit/js/DocumentUpdaterController.test.js index beb7c03563..7655ccd594 100644 --- a/services/real-time/test/unit/js/DocumentUpdaterController.test.js +++ b/services/real-time/test/unit/js/DocumentUpdaterController.test.js @@ -138,9 +138,9 @@ describe('DocumentUpdaterController', function () { ) }) - it('should apply the update', function (ctx) { + it('should apply the update to the doc room', function (ctx) { ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater - .calledWith(ctx.io, ctx.doc_id, ctx.message.op) + .calledWith(ctx.io, ctx.doc_id, ctx.doc_id, ctx.message.op) .should.equal(true) }) }) @@ -160,9 +160,89 @@ describe('DocumentUpdaterController', function () { ) }) - it('should process the error', function (ctx) { + it('should process the error in the doc room', function (ctx) { ctx.EditorUpdatesController._processErrorFromDocumentUpdater - .calledWith(ctx.io, ctx.doc_id, ctx.message.error) + .calledWith(ctx.io, ctx.doc_id, ctx.doc_id, ctx.message.error) + .should.equal(true) + }) + }) + }) + + describe('handleAppliedOpMessage', function () { + describe('with an update forwarded from the editor-events channel', function () { + beforeEach(function (ctx) { + ctx.message = { + project_id: ctx.project_id, + doc_id: ctx.doc_id, + op: { t: 'foo', p: 12 }, + } + ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater = + sinon.stub() + ctx.EditorUpdatesController.handleAppliedOpMessage( + ctx.io, + ctx.message, + ctx.project_id + ) + }) + + it('should apply the update to the project room', function (ctx) { + ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater + .calledWith(ctx.io, ctx.project_id, ctx.doc_id, ctx.message.op) + .should.equal(true) + }) + }) + + describe('with a duplicate update', function () { + beforeEach(function (ctx) { + ctx.settings.checkEventOrder = true + ctx.message = { + project_id: ctx.project_id, + doc_id: ctx.doc_id, + op: { t: 'foo', p: 12 }, + _id: 'doc:host:rnd-1', + } + ctx.EventLogger.checkEventOrder.returns('duplicate') + ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater = + sinon.stub() + ctx.EditorUpdatesController.handleAppliedOpMessage( + ctx.io, + ctx.message, + ctx.project_id + ) + }) + + it('should check the event order on the applied-ops channel', function (ctx) { + ctx.EventLogger.checkEventOrder + .calledWith('applied-ops', ctx.message._id, ctx.message) + .should.equal(true) + }) + + it('should skip the update', function (ctx) { + ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater.called.should.equal( + false + ) + }) + }) + + describe('with an error forwarded from the editor-events channel', function () { + beforeEach(function (ctx) { + ctx.message = { + project_id: ctx.project_id, + doc_id: ctx.doc_id, + error: 'Something went wrong', + } + ctx.EditorUpdatesController._processErrorFromDocumentUpdater = + sinon.stub() + ctx.EditorUpdatesController.handleAppliedOpMessage( + ctx.io, + ctx.message, + ctx.project_id + ) + }) + + it('should process the error in the project room', function (ctx) { + ctx.EditorUpdatesController._processErrorFromDocumentUpdater + .calledWith(ctx.io, ctx.project_id, ctx.doc_id, ctx.message.error) .should.equal(true) }) }) @@ -194,6 +274,7 @@ describe('DocumentUpdaterController', function () { ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater( ctx.io, ctx.doc_id, + ctx.doc_id, ctx.update ) }) @@ -224,6 +305,7 @@ describe('DocumentUpdaterController', function () { ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater( ctx.io, ctx.doc_id, + ctx.doc_id, ctx.update ) }) @@ -240,28 +322,95 @@ describe('DocumentUpdaterController', function () { ) }) }) + + describe('when broadcasting to the project room', function () { + beforeEach(function (ctx) { + ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater( + ctx.io, + ctx.project_id, + ctx.doc_id, + ctx.update + ) + }) + + it('should get the clients connected to the project', function (ctx) { + ctx.io.sockets.clients.calledWith(ctx.project_id).should.equal(true) + }) + + it('should send a version bump to the source client', function (ctx) { + ctx.sourceClient.emit + .calledWith('otUpdateApplied', { v: ctx.version, doc: ctx.doc_id }) + .should.equal(true) + ctx.sourceClient.emit.calledOnce.should.equal(true) + }) + + it('should send the full update to the other clients', function (ctx) { + Array.from(ctx.otherClients).map(client => + client.emit + .calledWith('otUpdateApplied', ctx.update) + .should.equal(true) + ) + }) + }) }) describe('_processErrorFromDocumentUpdater', function () { beforeEach(function (ctx) { ctx.clients = [new MockClient(), new MockClient()] - ctx.io.sockets = { clients: sinon.stub().returns(ctx.clients) } - ctx.EditorUpdatesController._processErrorFromDocumentUpdater( - ctx.io, - ctx.doc_id, - 'Something went wrong' - ) + for (const client of ctx.clients) { + client.ol_context[`doc:${ctx.doc_id}`] = 'allowed' + } }) - it('should log a warning', function (ctx) { - ctx.logger.warn.called.should.equal(true) + describe('in the doc room', function () { + beforeEach(function (ctx) { + ctx.io.sockets = { clients: sinon.stub().returns(ctx.clients) } + ctx.EditorUpdatesController._processErrorFromDocumentUpdater( + ctx.io, + ctx.doc_id, + ctx.doc_id, + 'Something went wrong' + ) + }) + + it('should log a warning', function (ctx) { + ctx.logger.warn.called.should.equal(true) + }) + + it('should disconnect all clients in that document', function (ctx) { + ctx.io.sockets.clients.calledWith(ctx.doc_id).should.equal(true) + Array.from(ctx.clients).map(client => + client.disconnect.called.should.equal(true) + ) + }) }) - it('should disconnect all clients in that document', function (ctx) { - ctx.io.sockets.clients.calledWith(ctx.doc_id).should.equal(true) - Array.from(ctx.clients).map(client => - client.disconnect.called.should.equal(true) - ) + describe('in the project room', function () { + beforeEach(function (ctx) { + ctx.otherDocClient = new MockClient() + ctx.otherDocClient.ol_context['doc:other-doc-id'] = 'allowed' + ctx.io.sockets = { + clients: sinon.stub().returns([...ctx.clients, ctx.otherDocClient]), + } + ctx.EditorUpdatesController._processErrorFromDocumentUpdater( + ctx.io, + ctx.project_id, + ctx.doc_id, + 'Something went wrong' + ) + }) + + it('should disconnect the clients that joined the doc', function (ctx) { + ctx.io.sockets.clients.calledWith(ctx.project_id).should.equal(true) + Array.from(ctx.clients).map(client => + client.disconnect.called.should.equal(true) + ) + }) + + it('should not disconnect clients that did not join the doc', function (ctx) { + ctx.otherDocClient.emit.called.should.equal(false) + ctx.otherDocClient.disconnect.called.should.equal(false) + }) }) }) }) diff --git a/services/real-time/test/unit/js/WebsocketLoadBalancer.test.js b/services/real-time/test/unit/js/WebsocketLoadBalancer.test.js index ca291ad513..8287676454 100644 --- a/services/real-time/test/unit/js/WebsocketLoadBalancer.test.js +++ b/services/real-time/test/unit/js/WebsocketLoadBalancer.test.js @@ -53,6 +53,12 @@ describe('WebsocketLoadBalancer', function () { }), })) + vi.doMock('../../../app/js/DocumentUpdaterController', () => ({ + default: (ctx.DocumentUpdaterController = { + handleAppliedOpMessage: sinon.stub(), + }), + })) + ctx.WebsocketLoadBalancer = (await import(modulePath)).default ctx.io = {} ctx.WebsocketLoadBalancer.rclientPubList = [{ publish: sinon.stub() }] @@ -64,7 +70,9 @@ describe('WebsocketLoadBalancer', function () { ] ctx.room_id = 'room-id' - ctx.message = 'otUpdateApplied' + // any non-restricted message name (must be on the restricted-user message + // pass list so the restricted-client cases receive it) + ctx.message = 'projectNameUpdated' ctx.payload = ['argument one', 42] }) @@ -337,8 +345,91 @@ describe('WebsocketLoadBalancer', function () { .should.equal(true) ctx.emit3.called.should.equal(false) }) + + it('should not forward the message to the DocumentUpdaterController', function (ctx) { + ctx.DocumentUpdaterController.handleAppliedOpMessage.called.should.equal( + false + ) + }) }) // duplicate client should be ignored + describe('with an applied op from document-updater', function () { + beforeEach(function (ctx) { + ctx.appliedOpMessage = { + project_id: 'project-id-123', + doc_id: 'doc-id-123', + op: { v: 42, doc: 'doc-id-123' }, + _id: 'doc:host:rnd-1', + message: 'otUpdateApplied', + } + ctx.io.sockets = { clients: sinon.stub().returns([]) } + ctx.WebsocketLoadBalancer._processEditorEvent( + ctx.io, + 'editor-events:project-id-123', + JSON.stringify(ctx.appliedOpMessage) + ) + }) + + it('should forward the message for broadcasting to the project room', function (ctx) { + ctx.DocumentUpdaterController.handleAppliedOpMessage + .calledWith(ctx.io, ctx.appliedOpMessage, 'project-id-123') + .should.equal(true) + }) + + it('should not emit the message as an editor event', function (ctx) { + ctx.io.sockets.clients.called.should.equal(false) + }) + }) + + describe('with an applied-op error from document-updater', function () { + beforeEach(function (ctx) { + ctx.appliedOpMessage = { + project_id: 'project-id-123', + doc_id: 'doc-id-123', + error: 'Something went wrong', + message: 'otUpdateError', + } + ctx.io.sockets = { clients: sinon.stub().returns([]) } + ctx.WebsocketLoadBalancer._processEditorEvent( + ctx.io, + 'editor-events:project-id-123', + JSON.stringify(ctx.appliedOpMessage) + ) + }) + + it('should forward the message for broadcasting to the project room', function (ctx) { + ctx.DocumentUpdaterController.handleAppliedOpMessage + .calledWith(ctx.io, ctx.appliedOpMessage, 'project-id-123') + .should.equal(true) + }) + }) + + describe('with an applied op without a project_id', function () { + beforeEach(function (ctx) { + ctx.appliedOpMessage = { + doc_id: 'doc-id-123', + op: { v: 42, doc: 'doc-id-123' }, + message: 'otUpdateApplied', + } + ctx.io.sockets = { clients: sinon.stub().returns([]) } + ctx.WebsocketLoadBalancer._processEditorEvent( + ctx.io, + 'editor-events', + JSON.stringify(ctx.appliedOpMessage) + ) + }) + + it('should log an error', function (ctx) { + ctx.logger.error.called.should.equal(true) + }) + + it('should not forward the message', function (ctx) { + ctx.DocumentUpdaterController.handleAppliedOpMessage.called.should.equal( + false + ) + }) + }) + describe('with a designated room, and restricted clients, not restricted message', function () { beforeEach(function (ctx) { ctx.io.sockets = {