[history-ot] initial implementation of using doc-level history-ot (#25054)

* [history-v1-ot] initial implementation of using doc-level history-v1-ot

* [web] fix advancing of the otMigrationStage

Use 'nextStage' for the user provided, desired stage when advancing.

Co-authored-by: Brian Gough <brian.gough@overleaf.com>

* [document-updater] document size check in editor-core

* [history-ot] rename history-v1-ot to history-ot and add types

* [history-ot] apply review feedback

- remove extra !!
- merge variable assignment when processing diff-match-match output
- add helper function for getting docstore lines view of StringFileData

Co-authored-by: Alf Eaton <alf.eaton@overleaf.com>

* Revert "[document-updater] add safe rollback point for history-ot (#25283)"

This reverts commit d7230dd14a379a27d2c6ab03a006463a18979d06

Signed-off-by: Jakob Ackermann <jakob.ackermann@overleaf.com>

---------

Signed-off-by: Jakob Ackermann <jakob.ackermann@overleaf.com>
Co-authored-by: Brian Gough <brian.gough@overleaf.com>
Co-authored-by: Alf Eaton <alf.eaton@overleaf.com>
GitOrigin-RevId: 89c497782adb0427635d50d02263d6f535b12481

services/real-time/app/js/ConnectedUsersManager.js

@@ -29,6 +29,10 @@ function recordProjectNotEmptySinceMetric(res, status) {
 }
 
 module.exports = {
+  countConnectedClients(projectId, callback) {
+    rclient.scard(Keys.clientsInProject({ project_id: projectId }), callback)
+  },
+
   // Use the same method for when a user connects, and when a user sends a cursor
   // update. This way we don't care if the connected_user key has expired when
   // we receive a cursor update.

services/real-time/app/js/DocumentUpdaterManager.js

@@ -19,7 +19,7 @@ const Keys = settings.redis.documentupdater.key_schema
 const DocumentUpdaterManager = {
   getDocument(projectId, docId, fromVersion, callback) {
     const timer = new metrics.Timer('get-document')
-    const url = `${settings.apis.documentupdater.url}/project/${projectId}/doc/${docId}?fromVersion=${fromVersion}`
+    const url = `${settings.apis.documentupdater.url}/project/${projectId}/doc/${docId}?fromVersion=${fromVersion}&historyV1OTSupport=true`
     logger.debug(
       { projectId, docId, fromVersion },
       'getting doc from document updater'
@@ -48,7 +48,8 @@ const DocumentUpdaterManager = {
           body.version,
           body.ranges,
           body.ops,
-          body.ttlInS
+          body.ttlInS,
+          body.type
         )
       } else if (res.statusCode === 422 && body?.firstVersionInRedis) {
         callback(new ClientRequestedMissingOpsError(422, body))

services/real-time/app/js/HttpApiController.js

@@ -1,8 +1,23 @@
 const WebsocketLoadBalancer = require('./WebsocketLoadBalancer')
 const DrainManager = require('./DrainManager')
+const ConnectedUsersManager = require('./ConnectedUsersManager')
 const logger = require('@overleaf/logger')
 
 module.exports = {
+  countConnectedClients(req, res) {
+    const { projectId } = req.params
+    ConnectedUsersManager.countConnectedClients(
+      projectId,
+      (err, nConnectedClients) => {
+        if (err) {
+          logger.err({ err, projectId }, 'count connected clients failed')
+          return res.sendStatus(500)
+        }
+        res.json({ nConnectedClients })
+      }
+    )
+  },
+
   sendMessage(req, res) {
     logger.debug({ message: req.params.message }, 'sending message')
     if (Array.isArray(req.body)) {

services/real-time/app/js/Router.js

@@ -113,6 +113,10 @@ module.exports = Router = {
       bodyParser.json({ limit: '5mb' }),
       HttpApiController.sendMessage
     )
+    app.get(
+      '/project/:projectId/count-connected-clients',
+      HttpApiController.countConnectedClients
+    )
 
     app.post('/drain', HttpApiController.startDrain)
     app.post(

services/real-time/app/js/WebsocketController.js

@@ -8,6 +8,7 @@ const ConnectedUsersManager = require('./ConnectedUsersManager')
 const WebsocketLoadBalancer = require('./WebsocketLoadBalancer')
 const RoomManager = require('./RoomManager')
 const {
+  CodedError,
   JoinLeaveEpochMismatchError,
   NotAuthorizedError,
   NotJoinedError,
@@ -283,7 +284,7 @@ module.exports = WebsocketController = {
             projectId,
             docId,
             fromVersion,
-            function (error, lines, version, ranges, ops, ttlInS) {
+            function (error, lines, version, ranges, ops, ttlInS, type) {
               if (error) {
                 if (error instanceof ClientRequestedMissingOpsError) {
                   emitJoinDocCatchUpMetrics('missing', error.info)
@@ -307,36 +308,53 @@ module.exports = WebsocketController = {
               // See http://ecmanaut.blogspot.co.uk/2006/07/encoding-decoding-utf8-in-javascript.html
               const encodeForWebsockets = text =>
                 unescape(encodeURIComponent(text))
-              const escapedLines = []
-              for (let line of lines) {
-                try {
-                  line = encodeForWebsockets(line)
-                } catch (err) {
-                  OError.tag(err, 'error encoding line uri component', { line })
-                  return callback(err)
+              metrics.inc('client_supports_history_v1_ot', 1, {
+                status: options.supportsHistoryV1OT ? 'success' : 'failure',
+              })
+              let escapedLines
+              if (type === 'history-ot') {
+                if (!options.supportsHistoryV1OT) {
+                  RoomManager.leaveDoc(client, docId)
+                  // TODO(24596): ask the user to reload the editor page (via out-of-sync modal when there are pending ops).
+                  return callback(
+                    new CodedError('client does not support history-ot')
+                  )
                 }
-                escapedLines.push(line)
-              }
-              if (options.encodeRanges) {
-                try {
-                  for (const comment of (ranges && ranges.comments) || []) {
-                    if (comment.op.c) {
-                      comment.op.c = encodeForWebsockets(comment.op.c)
-                    }
+                escapedLines = lines
+              } else {
+                escapedLines = []
+                for (let line of lines) {
+                  try {
+                    line = encodeForWebsockets(line)
+                  } catch (err) {
+                    OError.tag(err, 'error encoding line uri component', {
+                      line,
+                    })
+                    return callback(err)
                   }
-                  for (const change of (ranges && ranges.changes) || []) {
-                    if (change.op.i) {
-                      change.op.i = encodeForWebsockets(change.op.i)
+                  escapedLines.push(line)
+                }
+                if (options.encodeRanges) {
+                  try {
+                    for (const comment of (ranges && ranges.comments) || []) {
+                      if (comment.op.c) {
+                        comment.op.c = encodeForWebsockets(comment.op.c)
+                      }
                     }
-                    if (change.op.d) {
-                      change.op.d = encodeForWebsockets(change.op.d)
+                    for (const change of (ranges && ranges.changes) || []) {
+                      if (change.op.i) {
+                        change.op.i = encodeForWebsockets(change.op.i)
+                      }
+                      if (change.op.d) {
+                        change.op.d = encodeForWebsockets(change.op.d)
+                      }
                     }
+                  } catch (err) {
+                    OError.tag(err, 'error encoding range uri component', {
+                      ranges,
+                    })
+                    return callback(err)
                   }
-                } catch (err) {
-                  OError.tag(err, 'error encoding range uri component', {
-                    ranges,
-                  })
-                  return callback(err)
                 }
               }
 
@@ -351,7 +369,7 @@ module.exports = WebsocketController = {
                 },
                 'client joined doc'
               )
-              callback(null, escapedLines, version, ops, ranges)
+              callback(null, escapedLines, version, ops, ranges, type)
             }
           )
         })

services/real-time/test/acceptance/js/ClientTrackingTests.js

@@ -19,6 +19,80 @@ const FixturesManager = require('./helpers/FixturesManager')
 const async = require('async')
 
 describe('clientTracking', function () {
+  describe('when another logged in user joins a project', function () {
+    before(function (done) {
+      return async.series(
+        [
+          cb => {
+            return FixturesManager.setUpProject(
+              {
+                privilegeLevel: 'owner',
+                project: { name: 'Test Project' },
+              },
+              (error, { user_id: userId, project_id: projectId }) => {
+                if (error) return done(error)
+                this.user_id = userId
+                this.project_id = projectId
+                return cb()
+              }
+            )
+          },
+
+          cb => {
+            return FixturesManager.setUpDoc(
+              this.project_id,
+              { lines: this.lines, version: this.version, ops: this.ops },
+              (e, { doc_id: docId }) => {
+                this.doc_id = docId
+                return cb(e)
+              }
+            )
+          },
+
+          cb => {
+            this.clientA = RealTimeClient.connect(this.project_id, cb)
+          },
+
+          cb => {
+            RealTimeClient.countConnectedClients(
+              this.project_id,
+              (err, body) => {
+                if (err) return cb(err)
+                expect(body).to.deep.equal({ nConnectedClients: 1 })
+                cb()
+              }
+            )
+          },
+
+          cb => {
+            this.clientB = RealTimeClient.connect(this.project_id, cb)
+          },
+        ],
+        done
+      )
+    })
+
+    it('should record the initial state in getConnectedUsers', function (done) {
+      this.clientA.emit('clientTracking.getConnectedUsers', (error, users) => {
+        if (error) return done(error)
+        for (const user of Array.from(users)) {
+          if (user.client_id === this.clientB.publicId) {
+            expect(user.cursorData).to.not.exist
+            return done()
+          }
+        }
+        throw new Error('other user was never found')
+      })
+    })
+    it('should list both clients via HTTP', function (done) {
+      RealTimeClient.countConnectedClients(this.project_id, (err, body) => {
+        if (err) return done(err)
+        expect(body).to.deep.equal({ nConnectedClients: 2 })
+        done()
+      })
+    })
+  })
+
   describe('when a client updates its cursor location', function () {
     before(function (done) {
       return async.series(

services/real-time/test/acceptance/js/JoinDocTests.js

@@ -89,6 +89,7 @@ describe('joinDoc', function () {
         this.version,
         this.ops,
         this.ranges,
+        'sharejs-text-ot',
       ])
     })
 
@@ -168,6 +169,7 @@ describe('joinDoc', function () {
         this.version,
         this.ops,
         this.ranges,
+        'sharejs-text-ot',
       ])
     })
 
@@ -247,6 +249,7 @@ describe('joinDoc', function () {
         this.version,
         this.ops,
         this.ranges,
+        'sharejs-text-ot',
       ])
     })
 
@@ -408,6 +411,7 @@ describe('joinDoc', function () {
         this.version,
         this.ops,
         this.ranges,
+        'sharejs-text-ot',
       ])
     })
 
@@ -489,6 +493,7 @@ describe('joinDoc', function () {
         this.version,
         this.ops,
         this.ranges,
+        'sharejs-text-ot',
       ])
     })
 
@@ -504,7 +509,7 @@ describe('joinDoc', function () {
     })
   })
 
-  return describe('with fromVersion and options', function () {
+  describe('with fromVersion and options', function () {
     before(function (done) {
       this.fromVersion = 36
       this.options = { encodeRanges: true }
@@ -572,6 +577,7 @@ describe('joinDoc', function () {
         this.version,
         this.ops,
         this.ranges,
+        'sharejs-text-ot',
       ])
     })
 
@@ -586,4 +592,139 @@ describe('joinDoc', function () {
       )
     })
   })
+
+  describe('with type=history-ot', function () {
+    before(function (done) {
+      async.series(
+        [
+          cb => {
+            FixturesManager.setUpProject(
+              { privilegeLevel: 'owner' },
+              (e, { project_id: projectId, user_id: userId }) => {
+                this.project_id = projectId
+                this.user_id = userId
+                cb(e)
+              }
+            )
+          },
+
+          cb => {
+            FixturesManager.setUpDoc(
+              this.project_id,
+              {
+                lines: this.lines,
+                version: this.version,
+                ops: this.ops,
+                ranges: this.ranges,
+                type: 'history-ot',
+              },
+              (e, { doc_id: docId }) => {
+                this.doc_id = docId
+                cb(e)
+              }
+            )
+          },
+        ],
+        done
+      )
+    })
+
+    describe('when support is indicated', function () {
+      before(function (done) {
+        MockDocUpdaterServer.getDocument.resetHistory()
+        async.series(
+          [
+            cb => {
+              this.client = RealTimeClient.connect(this.project_id, cb)
+            },
+            cb =>
+              this.client.emit(
+                'joinDoc',
+                this.doc_id,
+                { supportsHistoryV1OT: true },
+                (error, ...rest) => {
+                  ;[...this.returnedArgs] = Array.from(rest)
+                  cb(error)
+                }
+              ),
+          ],
+          done
+        )
+      })
+
+      it('should get the doc from the doc updater', function () {
+        MockDocUpdaterServer.getDocument
+          .calledWith(this.project_id, this.doc_id, -1)
+          .should.equal(true)
+      })
+
+      it('should return the doc lines, version, ranges and ops', function () {
+        this.returnedArgs.should.deep.equal([
+          this.lines,
+          this.version,
+          this.ops,
+          this.ranges,
+          'history-ot',
+        ])
+      })
+
+      it('should have joined the doc room', function (done) {
+        RealTimeClient.getConnectedClient(
+          this.client.socket.sessionid,
+          (error, client) => {
+            if (error) return done(error)
+            expect(client.rooms).to.deep.equal([this.project_id, this.doc_id])
+            done()
+          }
+        )
+      })
+    })
+
+    describe('when support is not indicated', function () {
+      before(function (done) {
+        MockDocUpdaterServer.getDocument.resetHistory()
+        async.series(
+          [
+            cb => {
+              this.client = RealTimeClient.connect(this.project_id, cb)
+            },
+            cb =>
+              this.client.emit('joinDoc', this.doc_id, (error, ...rest) => {
+                this.error = error
+                ;[...this.returnedArgs] = Array.from(rest)
+                cb()
+              }),
+          ],
+          done
+        )
+      })
+
+      it('should get the doc from the doc updater', function () {
+        MockDocUpdaterServer.getDocument
+          .calledWith(this.project_id, this.doc_id, -1)
+          .should.equal(true)
+      })
+
+      it('should return an error', function () {
+        expect(this.error).to.deep.equal({
+          message: 'client does not support history-ot',
+        })
+      })
+
+      it('should not return the doc lines, version, ranges and ops', function () {
+        this.returnedArgs.should.deep.equal([])
+      })
+
+      it('should leave the doc room again', function (done) {
+        RealTimeClient.getConnectedClient(
+          this.client.socket.sessionid,
+          (error, client) => {
+            if (error) return done(error)
+            expect(client.rooms).to.deep.equal([this.project_id])
+            done()
+          }
+        )
+      })
+    })
+  })
 })

services/real-time/test/acceptance/js/helpers/FixturesManager.js

@@ -108,13 +108,17 @@ module.exports = FixturesManager = {
     if (!options.ops) {
       options.ops = ['mock', 'ops']
     }
-    const { doc_id: docId, lines, version, ops, ranges } = options
+    if (!options.type) {
+      options.type = 'sharejs-text-ot'
+    }
+    const { doc_id: docId, lines, version, ops, ranges, type } = options
 
     MockDocUpdaterServer.createMockDoc(projectId, docId, {
       lines,
       version,
       ops,
       ranges,
+      type,
     })
     return MockDocUpdaterServer.run(error => {
       if (error != null) {

services/real-time/test/acceptance/js/helpers/RealTimeClient.js

@@ -123,6 +123,16 @@ module.exports = Client = {
     )
   },
 
+  countConnectedClients(projectId, callback) {
+    request.get(
+      {
+        url: `http://127.0.0.1:3026/project/${projectId}/count-connected-clients`,
+        json: true,
+      },
+      (error, response, data) => callback(error, data)
+    )
+  },
+
   getConnectedClient(clientId, callback) {
     if (callback == null) {
       callback = function () {}

services/real-time/test/unit/js/DocumentUpdaterManagerTests.js

@@ -79,7 +79,7 @@ describe('DocumentUpdaterManager', function () {
       })
 
       it('should get the document from the document updater', function () {
-        const url = `${this.settings.apis.documentupdater.url}/project/${this.project_id}/doc/${this.doc_id}?fromVersion=${this.fromVersion}`
+        const url = `${this.settings.apis.documentupdater.url}/project/${this.project_id}/doc/${this.doc_id}?fromVersion=${this.fromVersion}&historyV1OTSupport=true`
         return this.request.get.calledWith(url).should.equal(true)
       })