diff --git a/services/document-updater/app/js/RealTimeRedisManager.js b/services/document-updater/app/js/RealTimeRedisManager.js index 5badc79246..2c03523b6d 100644 --- a/services/document-updater/app/js/RealTimeRedisManager.js +++ b/services/document-updater/app/js/RealTimeRedisManager.js @@ -106,48 +106,17 @@ const RealTimeRedisManager = { ) }, - /** - * Publish a canary message on the project's editor-events channel with the - * sizes of the applied op's ack and broadcast payloads. - * - * @param {{projectId: string, docId: string, op: Object}} args - */ - sendCanaryAppliedOp({ projectId, docId, op }) { - const ack = JSON.stringify({ v: op.v, doc: docId }).length - // Updates with op.dup===true will not get sent to other clients, they only get acked. - const broadcast = op.dup ? 0 : JSON.stringify(op).length - - const payload = JSON.stringify({ - message: 'canary-applied-op', - payload: { - ack, - broadcast, - docId, - projectId, - source: op.meta.source, - }, - }) - - // Publish on the editor-events channel of the project as real-time already listens to that before completing the connection startup. - - // publish on separate channels for individual projects and docs when - // configured (needs realtime to be configured for this too). - if (Settings.publishOnIndividualChannels) { - return pubsubClient.publish(`editor-events:${projectId}`, payload) - } else { - return pubsubClient.publish('editor-events', payload) - } - }, - /** * Publish an applied op or an error for a doc back to real-time. * + * The message is published on the project's editor-events channel, which + * real-time subscribes to for the lifetime of every connected project. It is + * broadcast to the project room in real-time; clients filter by doc id. + * * @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 + * editor-events channel and to the project room in real-time + * @param {string} data.doc_id - identifies the doc for clients * @param {Object} [data.op] - the applied op * @param {string} [data.error] - the error message when applying failed */ @@ -156,39 +125,26 @@ const RealTimeRedisManager = { const messageId = `doc:${HOST}:${RND}-${COUNT++}` 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' - } + // 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, { path }) + metrics.summary('redis.publish.applied-ops', blob.length, { + path: 'editor-events', + }) - // 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) - } - } + // publish on a per-project channel when configured (needs realtime to be + // configured for this too), otherwise on the base editor-events channel. if (Settings.publishOnIndividualChannels) { - return pubsubClient.publish(`applied-ops:${data.doc_id}`, blob) + return pubsubClient.publish(`editor-events:${data.project_id}`, blob) } else { - return pubsubClient.publish('applied-ops', blob) + return pubsubClient.publish('editor-events', blob) } }, } module.exports = RealTimeRedisManager module.exports.promises = promisifyAll(RealTimeRedisManager, { - without: ['sendCanaryAppliedOp', 'sendData'], + without: ['sendData'], }) diff --git a/services/document-updater/app/js/ShareJsUpdateManager.js b/services/document-updater/app/js/ShareJsUpdateManager.js index 933ded1ce1..99f70843db 100644 --- a/services/document-updater/app/js/ShareJsUpdateManager.js +++ b/services/document-updater/app/js/ShareJsUpdateManager.js @@ -133,11 +133,6 @@ const ShareJsUpdateManager = { doc_id: docId, op, }) - RealTimeRedisManager.sendCanaryAppliedOp({ - projectId, - docId, - op, - }) }, _computeHash(content) { diff --git a/services/document-updater/config/settings.defaults.js b/services/document-updater/config/settings.defaults.js index 2363de7255..6645c72c8a 100755 --- a/services/document-updater/config/settings.defaults.js +++ b/services/document-updater/config/settings.defaults.js @@ -185,16 +185,6 @@ 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 a502c4a9e3..cd6117caaf 100644 --- a/services/document-updater/test/acceptance/js/ApplyingUpdatesToADocTests.js +++ b/services/document-updater/test/acceptance/js/ApplyingUpdatesToADocTests.js @@ -649,7 +649,7 @@ describe('Applying updates to a doc', function () { it('should send a message with an error', function () { this.messageCallback.called.should.equal(true) const [channel, message] = this.messageCallback.args[0] - channel.should.equal('applied-ops') + channel.should.equal('editor-events') JSON.parse(message).should.deep.include({ project_id: this.project_id, doc_id: this.doc_id, @@ -658,82 +658,36 @@ describe('Applying updates to a doc', function () { }) }) - describe('with publishAppliedOpsOnEditorEvents enabled', function () { - beforeEach(function () { - Settings.publishAppliedOpsOnEditorEvents = true + 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) }) - afterEach(function () { - Settings.publishAppliedOpsOnEditorEvents = false + 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) }) - 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', - }) + 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) }) }) @@ -766,7 +720,7 @@ describe('Applying updates to a doc', function () { it('should send a message with an error', function () { this.messageCallback.called.should.equal(true) const [channel, message] = this.messageCallback.args[0] - channel.should.equal('applied-ops') + channel.should.equal('editor-events') JSON.parse(message).should.deep.include({ project_id: this.project_id, doc_id: this.doc_id, @@ -803,7 +757,7 @@ describe('Applying updates to a doc', function () { it('should send a message with an error', function () { this.messageCallback.called.should.equal(true) const [channel, message] = this.messageCallback.args[0] - channel.should.equal('applied-ops') + channel.should.equal('editor-events') JSON.parse(message).should.deep.include({ project_id: this.project_id, doc_id: this.doc_id, @@ -835,7 +789,7 @@ describe('Applying updates to a doc', function () { it('should send a message with an error', function () { this.messageCallback.called.should.equal(true) const [channel, message] = this.messageCallback.args[0] - channel.should.equal('applied-ops') + channel.should.equal('editor-events') JSON.parse(message).should.deep.include({ project_id: this.project_id, doc_id: this.doc_id, @@ -913,9 +867,9 @@ describe('Applying updates to a doc', function () { it('should return a message about duplicate ops', function () { this.messageCallback.calledTwice.should.equal(true) - this.messageCallback.args[0][0].should.equal('applied-ops') + this.messageCallback.args[0][0].should.equal('editor-events') expect(JSON.parse(this.messageCallback.args[0][1]).op.dup).to.be.undefined - this.messageCallback.args[1][0].should.equal('applied-ops') + this.messageCallback.args[1][0].should.equal('editor-events') expect(JSON.parse(this.messageCallback.args[1][1]).op.dup).to.equal(true) }) }) @@ -964,9 +918,9 @@ describe('Applying updates to a doc', function () { it('should return a message about duplicate ops', function () { this.messageCallback.calledTwice.should.equal(true) - this.messageCallback.args[0][0].should.equal('applied-ops') + this.messageCallback.args[0][0].should.equal('editor-events') expect(JSON.parse(this.messageCallback.args[0][1]).op.dup).to.be.undefined - this.messageCallback.args[1][0].should.equal('applied-ops') + this.messageCallback.args[1][0].should.equal('editor-events') expect(JSON.parse(this.messageCallback.args[1][1]).op.dup).to.equal(true) }) }) @@ -995,7 +949,7 @@ describe('Applying updates to a doc', function () { it('should send a message with an error', function () { this.messageCallback.called.should.equal(true) const [channel, message] = this.messageCallback.args[0] - channel.should.equal('applied-ops') + channel.should.equal('editor-events') JSON.parse(message).should.deep.include({ project_id: this.project_id, doc_id: this.doc_id, diff --git a/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js b/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js index bf68dfd80b..3c172ee495 100644 --- a/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js +++ b/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js @@ -11,9 +11,7 @@ const { setTimeout } = require('node:timers/promises') 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 +// applied ops (and errors) are published on the editor-events channel rclientSub.subscribe('editor-events') rclientSub.setMaxListeners(0) @@ -36,12 +34,7 @@ module.exports = DocUpdaterClient = { }, subscribeToAppliedOps(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) - }) + rclientSub.on('message', messageHandler) }, // 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 90544207d6..444b4aeb24 100644 --- a/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js +++ b/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js @@ -148,50 +148,15 @@ describe('RealTimeRedisManager', function () { doc_id: this.doc_id, op: 'thisop', } - this.blob = JSON.stringify({ ...this.data, _id: this.message_id }) - }) - - 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) - }) - }) - - 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) + this.blob = JSON.stringify({ + ...this.data, + _id: this.message_id, + message: 'otUpdateApplied', }) }) 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) }) @@ -212,13 +177,7 @@ describe('RealTimeRedisManager', function () { 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) }) @@ -229,9 +188,8 @@ describe('RealTimeRedisManager', function () { }) }) - describe('with an error on the editor-events channel', function () { + describe('with an error', function () { beforeEach(function () { - this.Settings.publishAppliedOpsOnEditorEvents = true this.data = { project_id: this.project_id, doc_id: this.doc_id, diff --git a/services/document-updater/test/unit/js/ShareJsUpdateManager/ShareJsUpdateManagerTests.js b/services/document-updater/test/unit/js/ShareJsUpdateManager/ShareJsUpdateManagerTests.js index e699439a74..68aec9e908 100644 --- a/services/document-updater/test/unit/js/ShareJsUpdateManager/ShareJsUpdateManagerTests.js +++ b/services/document-updater/test/unit/js/ShareJsUpdateManager/ShareJsUpdateManagerTests.js @@ -33,9 +33,7 @@ describe('ShareJsUpdateManager', function () { return (this.rclient = { auth() {} }) }, }, - './RealTimeRedisManager': (this.RealTimeRedisManager = { - sendCanaryAppliedOp: sinon.stub(), - }), + './RealTimeRedisManager': (this.RealTimeRedisManager = {}), './Metrics': (this.metrics = { inc: sinon.stub() }), }, globals: { diff --git a/services/real-time/app/js/WebsocketLoadBalancer.js b/services/real-time/app/js/WebsocketLoadBalancer.js index a6c4549c35..83e54ad55f 100644 --- a/services/real-time/app/js/WebsocketLoadBalancer.js +++ b/services/real-time/app/js/WebsocketLoadBalancer.js @@ -26,12 +26,6 @@ const RESTRICTED_USER_MESSAGE_TYPE_PASS_LIST = [ 'toggle-track-changes', 'projectRenamedOrDeletedByExternalSource', ] -const BANDWIDTH_BUCKETS = [0] -// 64 bytes ... 8MB -for (let i = 5; i <= 22; i++) { - BANDWIDTH_BUCKETS.push(2 << i) -} - let WebsocketLoadBalancer export default WebsocketLoadBalancer = { @@ -147,31 +141,6 @@ export default WebsocketLoadBalancer = { for (const client of clientList) { ConnectedUsersManager.refreshClient(message.room_id, client.publicId) } - } else if (message.message === 'canary-applied-op') { - const { ack, broadcast, source, projectId, docId } = message.payload - - const estimateBandwidth = (room, path) => { - const seen = new Set() - for (const client of io.sockets.clients(room)) { - if (seen.has(client.id)) continue - seen.add(client.id) - let v = client.id === source ? ack : broadcast - if (v === 0) { - // Acknowledgements with update.dup===true will not get sent to other clients. - continue - } - v += `5:::{"name":"otUpdateApplied","args":[]}`.length - Metrics.histogram( - 'estimated-applied-ops-bandwidth', - v, - BANDWIDTH_BUCKETS, - { path } - ) - } - } - - estimateBandwidth(projectId, 'per-project') - estimateBandwidth(docId, 'per-doc') } else if ( message.message === 'otUpdateApplied' || message.message === 'otUpdateError'