[monorepo] clean up the per-project pub/sub migration (#35686)

GitOrigin-RevId: 4bbb97d1d2750f21751ef54e206e3e720b88615a

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'],
 })

services/document-updater/app/js/ShareJsUpdateManager.js

@@ -133,11 +133,6 @@ const ShareJsUpdateManager = {
       doc_id: docId,
       op,
     })
-    RealTimeRedisManager.sendCanaryAppliedOp({
-      projectId,
-      docId,
-      op,
-    })
   },
 
   _computeHash(content) {

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

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,

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

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,

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: {

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'