[monorepo] prepare per-project queues (#35386)

GitOrigin-RevId: ab020603ec04c705d08be466563adac0ce30bc04

services/document-updater/test/acceptance/js/QueueMigrationTests.js

@@ -0,0 +1,181 @@
+/*
+ * Exercises the migration of the real-time -> document-updater update queue
+ * from the per-doc queue to the per-project queue.
+ *
+ * The consumer's behaviour is gated by Settings.pendingUpdatesMigrationPhase,
+ * which the app reads live. The acceptance app runs in-process, so each block
+ * sets the phase it needs in a `before` hook and the top-level `after` restores
+ * the original value.
+ *
+ * The dual-read consumer (phases 1 & 2) must drain BOTH the legacy per-doc queue
+ * and the new per-project queue, so updates queued either side of a producer
+ * roll-forward/roll-backward are still applied. Ordering is relaxed (ShareJS
+ * rebases by version); these tests keep versions monotonic across the switch.
+ */
+const { expect } = require('chai')
+const { setTimeout } = require('node:timers/promises')
+const Settings = require('@overleaf/settings')
+const rclientDU = require('@overleaf/redis-wrapper').createClient(
+  Settings.redis.documentupdater
+)
+const Keys = Settings.redis.documentupdater.key_schema
+
+const MockWebApi = require('./helpers/MockWebApi')
+const DocUpdaterClient = require('./helpers/DocUpdaterClient')
+const DocUpdaterApp = require('./helpers/DocUpdaterApp')
+
+function insertOp(text, position, version, docId) {
+  return { doc: docId, op: [{ i: text, p: position }], v: version }
+}
+
+describe('Queue migration', function () {
+  before(function () {
+    this.originalPhase = Settings.pendingUpdatesMigrationPhase
+  })
+
+  after(function () {
+    Settings.pendingUpdatesMigrationPhase = this.originalPhase
+  })
+
+  beforeEach(async function () {
+    this.lines = ['one', 'two', 'three']
+    this.version = 0
+    this.project_id = DocUpdaterClient.randomId()
+    this.doc_id = DocUpdaterClient.randomId()
+    MockWebApi.insertDoc(this.project_id, this.doc_id, {
+      lines: this.lines,
+      version: this.version,
+    })
+    await DocUpdaterApp.ensureRunning()
+    await DocUpdaterClient.preloadDoc(this.project_id, this.doc_id)
+  })
+
+  describe('dual-read consumer (phases 1 & 2)', function () {
+    before(function () {
+      Settings.pendingUpdatesMigrationPhase = 2
+    })
+
+    it('roll-forward: legacy updates then per-project updates all apply in order', async function () {
+      await DocUpdaterClient.sendUpdate(
+        this.project_id,
+        this.doc_id,
+        insertOp('AAA\n', 0, 0, this.doc_id),
+        { toProjectQueue: false }
+      )
+      await DocUpdaterClient.waitForPendingUpdates(this.project_id, this.doc_id)
+
+      await DocUpdaterClient.sendUpdate(
+        this.project_id,
+        this.doc_id,
+        insertOp('BBB\n', 0, 1, this.doc_id),
+        { toProjectQueue: true }
+      )
+      await DocUpdaterClient.waitForPendingUpdates(this.project_id, this.doc_id)
+
+      const doc = await DocUpdaterClient.getDoc(this.project_id, this.doc_id)
+      doc.lines.should.deep.equal(['BBB', 'AAA', 'one', 'two', 'three'])
+      doc.version.should.equal(2)
+    })
+
+    it('roll-backward: per-project updates then legacy updates all apply in order', async function () {
+      await DocUpdaterClient.sendUpdate(
+        this.project_id,
+        this.doc_id,
+        insertOp('AAA\n', 0, 0, this.doc_id),
+        { toProjectQueue: true }
+      )
+      await DocUpdaterClient.waitForPendingUpdates(this.project_id, this.doc_id)
+
+      await DocUpdaterClient.sendUpdate(
+        this.project_id,
+        this.doc_id,
+        insertOp('BBB\n', 0, 1, this.doc_id),
+        { toProjectQueue: false }
+      )
+      await DocUpdaterClient.waitForPendingUpdates(this.project_id, this.doc_id)
+
+      const doc = await DocUpdaterClient.getDoc(this.project_id, this.doc_id)
+      doc.lines.should.deep.equal(['BBB', 'AAA', 'one', 'two', 'three'])
+      doc.version.should.equal(2)
+    })
+
+    it('straddle: updates present in BOTH queues drain together (per-doc first), in order', async function () {
+      // Populate both queues before any dispatch, then trigger a single
+      // whole-project drain with a bare project-id marker.
+      await rclientDU.rpush(
+        Keys.pendingUpdates({ doc_id: this.doc_id }),
+        JSON.stringify(insertOp('AAA\n', 0, 0, this.doc_id))
+      )
+      await rclientDU.rpush(
+        Keys.pendingProjectUpdates({ project_id: this.project_id }),
+        JSON.stringify(insertOp('BBB\n', 0, 1, this.doc_id))
+      )
+      await rclientDU.rpush('pending-updates-list', this.project_id)
+
+      await DocUpdaterClient.waitForPendingUpdates(this.project_id, this.doc_id)
+      await setTimeout(200)
+
+      const doc = await DocUpdaterClient.getDoc(this.project_id, this.doc_id)
+      doc.lines.should.deep.equal(['BBB', 'AAA', 'one', 'two', 'three'])
+      doc.version.should.equal(2)
+    })
+
+    it('flush applies per-project queue updates first (no stale flush)', async function () {
+      // Queue an update on the per-project queue WITHOUT a dispatch marker, so
+      // it is only picked up by the flush path draining outstanding updates.
+      await rclientDU.rpush(
+        Keys.pendingProjectUpdates({ project_id: this.project_id }),
+        JSON.stringify(insertOp('AAA\n', 0, 0, this.doc_id))
+      )
+
+      await DocUpdaterClient.flushDoc(this.project_id, this.doc_id)
+
+      const doc = await DocUpdaterClient.getDoc(this.project_id, this.doc_id)
+      doc.lines.should.deep.equal(['AAA', 'one', 'two', 'three'])
+      doc.version.should.equal(1)
+      const remaining = await rclientDU.llen(
+        Keys.pendingProjectUpdates({ project_id: this.project_id })
+      )
+      expect(remaining).to.equal(0)
+    })
+  })
+
+  describe('project-queue-only consumer (phase 3)', function () {
+    before(function () {
+      Settings.pendingUpdatesMigrationPhase = 3
+    })
+
+    it('drains the per-project queue', async function () {
+      await DocUpdaterClient.sendUpdate(
+        this.project_id,
+        this.doc_id,
+        insertOp('AAA\n', 0, 0, this.doc_id),
+        { toProjectQueue: true }
+      )
+      await DocUpdaterClient.waitForPendingUpdates(this.project_id, this.doc_id)
+      const doc = await DocUpdaterClient.getDoc(this.project_id, this.doc_id)
+      doc.lines.should.deep.equal(['AAA', 'one', 'two', 'three'])
+      doc.version.should.equal(1)
+    })
+
+    it('ignores the legacy per-doc queue', async function () {
+      // Push to the legacy per-doc queue with its marker; the phase-3 consumer
+      // must NOT process it (it no longer reads the per-doc queue).
+      await DocUpdaterClient.sendUpdate(
+        this.project_id,
+        this.doc_id,
+        insertOp('AAA\n', 0, 0, this.doc_id),
+        { toProjectQueue: false }
+      )
+      await setTimeout(500)
+
+      const doc = await DocUpdaterClient.getDoc(this.project_id, this.doc_id)
+      doc.lines.should.deep.equal(this.lines)
+      doc.version.should.equal(0)
+      const remaining = await rclientDU.llen(
+        Keys.pendingUpdates({ doc_id: this.doc_id })
+      )
+      expect(remaining).to.equal(1)
+    })
+  })
+})

services/document-updater/test/acceptance/js/SettingADocumentTests.js

@@ -546,7 +546,7 @@ describe('Setting a document', function () {
         this.doc_id,
         this.historyOTUpdate
       )
-      await DocUpdaterClient.waitForPendingUpdates(this.doc_id)
+      await DocUpdaterClient.waitForPendingUpdates(this.project_id, this.doc_id)
     })
 
     afterEach(function () {

services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js

@@ -36,32 +36,52 @@ module.exports = DocUpdaterClient = {
     rclientSub.on('message', messageHandler)
   },
 
-  async sendUpdate(projectId, docId, update) {
-    const docKey = `${projectId}:${docId}`
-    await rclient.rpush(
-      keys.pendingUpdates({ doc_id: docId }),
-      JSON.stringify(update)
-    )
-    await rclient.sadd('DocsWithPendingUpdates', docKey)
-    await rclient.rpush(getPendingUpdateListKey(), docKey)
+  // Enqueue an update for the doc-updater. By default the queue format follows
+  // the configured migration phase, but tests can force a format with
+  // options.toProjectQueue to exercise the migration explicitly:
+  //   phase 1 / toProjectQueue=false: legacy per-doc queue + "<proj>:<doc>" marker
+  //   phase >= 2 / toProjectQueue=true: per-project queue + bare "<proj>" marker
+  async sendUpdate(projectId, docId, update, options = {}) {
+    const toProjectQueue =
+      options.toProjectQueue ?? Settings.pendingUpdatesMigrationPhase >= 2
+    if (toProjectQueue) {
+      await rclient.rpush(
+        keys.pendingProjectUpdates({ project_id: projectId }),
+        JSON.stringify({ ...update, doc: docId })
+      )
+      await rclient.rpush(getPendingUpdateListKey(), projectId)
+    } else {
+      await rclient.rpush(
+        keys.pendingUpdates({ doc_id: docId }),
+        JSON.stringify(update)
+      )
+      await rclient.rpush(getPendingUpdateListKey(), `${projectId}:${docId}`)
+    }
   },
 
-  async sendUpdates(projectId, docId, updates) {
+  async sendUpdates(projectId, docId, updates, options = {}) {
     await DocUpdaterClient.preloadDoc(projectId, docId)
     for (const update of updates) {
-      await DocUpdaterClient.sendUpdate(projectId, docId, update)
+      await DocUpdaterClient.sendUpdate(projectId, docId, update, options)
     }
-    await DocUpdaterClient.waitForPendingUpdates(docId)
+    await DocUpdaterClient.waitForPendingUpdates(projectId, docId)
   },
 
-  async waitForPendingUpdates(docId) {
+  // Wait until both the legacy per-doc queue and the per-project queue are
+  // drained, so the helper works regardless of the migration phase in use.
+  async waitForPendingUpdates(projectId, docId) {
     const maxRetries = 30
     const retryInterval = 100
 
     for (let attempt = 0; attempt < maxRetries; attempt++) {
-      const length = await rclient.llen(keys.pendingUpdates({ doc_id: docId }))
+      const docLength = await rclient.llen(
+        keys.pendingUpdates({ doc_id: docId })
+      )
+      const projectLength = await rclient.llen(
+        keys.pendingProjectUpdates({ project_id: projectId })
+      )
 
-      if (length === 0) {
+      if (docLength === 0 && projectLength === 0) {
         return // Success - no pending updates
       }
 

services/document-updater/test/unit/js/DispatchManager/DispatchManagerTests.js

@@ -139,6 +139,28 @@ describe('DispatchManager', function () {
         })
       })
 
+      describe('with a bare project-id marker (per-project queue)', function () {
+        beforeEach(function () {
+          this.client.blpop = sinon
+            .stub()
+            .callsArgWith(2, null, ['pending-updates-list', this.project_id])
+          this.UpdateManager.processOutstandingUpdatesWithLock = sinon
+            .stub()
+            .callsArg(2)
+          return this.worker._waitForUpdateThenDispatchWorker(this.callback)
+        })
+
+        it('should call processOutstandingUpdatesWithLock with no doc hint', function () {
+          this.UpdateManager.processOutstandingUpdatesWithLock
+            .calledWith(this.project_id, undefined)
+            .should.equal(true)
+        })
+
+        it('should call the callback', function () {
+          return this.callback.called.should.equal(true)
+        })
+      })
+
       describe('pending updates list with shard key', function () {
         beforeEach(function (done) {
           this.client = {

services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js

@@ -35,6 +35,9 @@ describe('RealTimeRedisManager', function () {
                 pendingUpdates({ doc_id: docId }) {
                   return `PendingUpdates:${docId}`
                 },
+                pendingProjectUpdates({ project_id: projectId }) {
+                  return `PendingProjectUpdates:${projectId}`
+                },
               },
             }),
             pubsub: {
@@ -140,6 +143,85 @@ describe('RealTimeRedisManager', function () {
     })
   })
 
+  describe('getPendingProjectUpdates', function () {
+    beforeEach(function () {
+      this.rclient.llen = sinon.stub()
+      this.rclient.lrange = sinon.stub()
+      this.rclient.ltrim = sinon.stub()
+    })
+
+    describe('successfully', function () {
+      beforeEach(function () {
+        this.updates = [
+          { doc: 'doc-1', op: [{ i: 'foo', p: 4 }] },
+          { doc: 'doc-2', op: [{ i: 'bar', p: 6 }] },
+        ]
+        this.jsonUpdates = this.updates.map(update => JSON.stringify(update))
+        this.rclient.exec = sinon.stub().yields(null, [2, this.jsonUpdates])
+        return this.RealTimeRedisManager.getPendingProjectUpdates(
+          this.project_id,
+          this.callback
+        )
+      })
+
+      it('should get the pending updates from the per-project queue', function () {
+        return this.rclient.lrange
+          .calledWith(`PendingProjectUpdates:${this.project_id}`, 0, 7)
+          .should.equal(true)
+      })
+
+      it('should delete the pending updates', function () {
+        return this.rclient.ltrim
+          .calledWith(`PendingProjectUpdates:${this.project_id}`, 8, -1)
+          .should.equal(true)
+      })
+
+      return it('should call the callback with the updates', function () {
+        return this.callback.calledWith(null, this.updates).should.equal(true)
+      })
+    })
+
+    return describe("when the JSON doesn't parse", function () {
+      beforeEach(function () {
+        this.jsonUpdates = [
+          JSON.stringify({ doc: 'doc-1', op: [{ i: 'foo', p: 4 }] }),
+          'broken json',
+        ]
+        this.rclient.exec = sinon.stub().yields(null, [2, this.jsonUpdates])
+        return this.RealTimeRedisManager.getPendingProjectUpdates(
+          this.project_id,
+          this.callback
+        )
+      })
+
+      return it('should return an error to the callback', function () {
+        return this.callback
+          .calledWith(sinon.match.has('name', 'SyntaxError'))
+          .should.equal(true)
+      })
+    })
+  })
+
+  describe('getProjectUpdatesLength', function () {
+    beforeEach(function () {
+      this.rclient.llen = sinon.stub().yields(null, (this.length = 4))
+      return this.RealTimeRedisManager.getProjectUpdatesLength(
+        this.project_id,
+        this.callback
+      )
+    })
+
+    it('should look up the length of the per-project queue', function () {
+      return this.rclient.llen
+        .calledWith(`PendingProjectUpdates:${this.project_id}`)
+        .should.equal(true)
+    })
+
+    return it('should return the length', function () {
+      return this.callback.calledWith(null, this.length).should.equal(true)
+    })
+  })
+
   return describe('sendData', function () {
     beforeEach(function () {
       this.message_id = 'doc:somehost:01020304-0'

services/document-updater/test/unit/js/UpdateManager/UpdateManagerTests.js

@@ -36,6 +36,7 @@ describe('UpdateManager', function () {
       promises: {
         tryLock: sinon.stub().resolves(this.projectLockValue),
         getLock: sinon.stub().resolves(this.projectLockValue),
+        extendLock: sinon.stub().resolves(),
         releaseLock: sinon.stub().resolves(),
       },
     }
@@ -45,17 +46,25 @@ describe('UpdateManager', function () {
         setDocument: sinon.stub().resolves(),
         updateDocument: sinon.stub(),
         recordProjectNotificationTimestamp: sinon.stub().resolves(),
+        getDocIdsInProject: sinon.stub().resolves([]),
       },
     }
 
     this.RealTimeRedisManager = {
       sendData: sinon.stub(),
       promises: {
-        getUpdatesLength: sinon.stub(),
+        getUpdatesLength: sinon.stub().resolves(0),
         getPendingUpdatesForDoc: sinon.stub(),
+        getProjectUpdatesLength: sinon.stub().resolves(0),
+        getPendingProjectUpdates: sinon.stub().resolves([]),
       },
     }
 
+    this.HistoryOTUpdateManager = {
+      isHistoryOTEditOperationUpdate: sinon.stub().returns(false),
+      applyUpdate: sinon.stub().resolves(),
+    }
+
     this.ShareJsUpdateManager = {
       promises: {
         applyUpdate: sinon.stub(),
@@ -66,7 +75,7 @@ describe('UpdateManager', function () {
       recordAndFlushHistoryOps: sinon.stub(),
     }
 
-    this.Settings = {}
+    this.Settings = { pendingUpdatesMigrationPhase: 1 }
 
     this.DocumentManager = {
       promises: {
@@ -106,6 +115,7 @@ describe('UpdateManager', function () {
         './RealTimeRedisManager': this.RealTimeRedisManager,
         './ShareJsUpdateManager': this.ShareJsUpdateManager,
         './HistoryManager': this.HistoryManager,
+        './HistoryOTUpdateManager': this.HistoryOTUpdateManager,
         './Metrics': this.Metrics,
         '@overleaf/settings': this.Settings,
         './DocumentManager': this.DocumentManager,
@@ -139,12 +149,12 @@ describe('UpdateManager', function () {
   })
 
   describe('processOutstandingUpdatesWithLock', function () {
-    describe('when the lock is free', function () {
+    describe('when the project lock is free', function () {
       beforeEach(function () {
         this.UpdateManager.promises.continueProcessingUpdatesWithLock = sinon
           .stub()
           .resolves()
-        this.UpdateManager.promises.processOutstandingUpdates = sinon
+        this.UpdateManager.promises.processOutstandingProjectUpdates = sinon
           .stub()
           .resolves()
       })
@@ -157,25 +167,10 @@ describe('UpdateManager', function () {
           )
         })
 
-        it('should acquire the project lock before the doc lock', function () {
+        it('should acquire the project lock', function () {
           this.ProjectLockManager.promises.getLock
             .calledWith(this.project_id)
             .should.equal(true)
-          this.LockManager.promises.tryLock
-            .calledAfter(this.ProjectLockManager.promises.getLock)
-            .should.equal(true)
-        })
-
-        it('should acquire the lock', function () {
-          this.LockManager.promises.tryLock
-            .calledWith(this.doc_id)
-            .should.equal(true)
-        })
-
-        it('should free the lock', function () {
-          this.LockManager.promises.releaseLock
-            .calledWith(this.doc_id, this.lockValue)
-            .should.equal(true)
         })
 
         it('should free the project lock', function () {
@@ -184,18 +179,18 @@ describe('UpdateManager', function () {
             .should.equal(true)
         })
 
-        it('should process the outstanding updates', function () {
-          this.UpdateManager.promises.processOutstandingUpdates
-            .calledWith(this.project_id, this.doc_id)
+        it('should process the outstanding project updates', function () {
+          this.UpdateManager.promises.processOutstandingProjectUpdates
+            .calledWith(this.project_id, this.doc_id, this.projectLockValue)
             .should.equal(true)
         })
 
         it('should do everything with the lock acquired', function () {
-          this.UpdateManager.promises.processOutstandingUpdates
-            .calledAfter(this.LockManager.promises.tryLock)
+          this.UpdateManager.promises.processOutstandingProjectUpdates
+            .calledAfter(this.ProjectLockManager.promises.getLock)
             .should.equal(true)
-          this.UpdateManager.promises.processOutstandingUpdates
-            .calledBefore(this.LockManager.promises.releaseLock)
+          this.UpdateManager.promises.processOutstandingProjectUpdates
+            .calledBefore(this.ProjectLockManager.promises.releaseLock)
             .should.equal(true)
         })
 
@@ -206,10 +201,10 @@ describe('UpdateManager', function () {
         })
       })
 
-      describe('when processOutstandingUpdates returns an error', function () {
+      describe('when processOutstandingProjectUpdates returns an error', function () {
         beforeEach(async function () {
           this.error = new Error('Something went wrong')
-          this.UpdateManager.promises.processOutstandingUpdates = sinon
+          this.UpdateManager.promises.processOutstandingProjectUpdates = sinon
             .stub()
             .rejects(this.error)
           await expect(
@@ -220,12 +215,6 @@ describe('UpdateManager', function () {
           ).to.be.rejectedWith(this.error)
         })
 
-        it('should free the lock', function () {
-          this.LockManager.promises.releaseLock
-            .calledWith(this.doc_id, this.lockValue)
-            .should.equal(true)
-        })
-
         it('should free the project lock', function () {
           this.ProjectLockManager.promises.releaseLock
             .calledWith(this.project_id, this.projectLockValue)
@@ -233,37 +222,247 @@ describe('UpdateManager', function () {
         })
       })
     })
+  })
 
-    describe('when the lock is taken', function () {
+  describe('processOutstandingProjectUpdates', function () {
+    beforeEach(function () {
+      this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock = sinon
+        .stub()
+        .resolves()
+      this.UpdateManager.promises.fetchAndApplyProjectUpdates = sinon
+        .stub()
+        .resolves()
+    })
+
+    describe('in phases 1 & 2', function () {
       beforeEach(async function () {
-        this.LockManager.promises.tryLock.resolves(null)
-        this.UpdateManager.promises.processOutstandingUpdates = sinon
-          .stub()
-          .resolves()
-        await this.UpdateManager.promises.processOutstandingUpdatesWithLock(
+        this.Settings.pendingUpdatesMigrationPhase = 2
+        this.RedisManager.promises.getDocIdsInProject.resolves([
+          'doc-1',
+          'doc-2',
+        ])
+        await this.UpdateManager.promises.processOutstandingProjectUpdates(
+          this.project_id,
+          this.doc_id,
+          this.projectLockValue,
+          new this.Profiler()
+        )
+      })
+
+      it('should drain the per-doc queue for every doc in the project plus the hint', function () {
+        this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock.should.have.been.calledWith(
+          this.project_id,
+          'doc-1'
+        )
+        this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock.should.have.been.calledWith(
+          this.project_id,
+          'doc-2'
+        )
+        this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock.should.have.been.calledWith(
           this.project_id,
           this.doc_id
         )
       })
 
-      it('should not process the updates', function () {
-        this.UpdateManager.promises.processOutstandingUpdates.called.should.equal(
+      it('should extend the project lock between docs', function () {
+        this.ProjectLockManager.promises.extendLock.should.have.been.calledWith(
+          this.project_id,
+          this.projectLockValue
+        )
+      })
+
+      it('should drain the per-project queue', function () {
+        this.UpdateManager.promises.fetchAndApplyProjectUpdates.should.have.been.calledWith(
+          this.project_id
+        )
+      })
+    })
+
+    describe('in phase 3', function () {
+      beforeEach(async function () {
+        this.Settings.pendingUpdatesMigrationPhase = 3
+        await this.UpdateManager.promises.processOutstandingProjectUpdates(
+          this.project_id,
+          this.doc_id,
+          this.projectLockValue,
+          new this.Profiler()
+        )
+      })
+
+      it('should not look up the docs in the project', function () {
+        this.RedisManager.promises.getDocIdsInProject.called.should.equal(false)
+      })
+
+      it('should not drain any per-doc queue', function () {
+        this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock.called.should.equal(
           false
         )
       })
 
-      it('should free the project lock', function () {
-        this.ProjectLockManager.promises.releaseLock
-          .calledWith(this.project_id, this.projectLockValue)
+      it('should not bump the project lock (freshly acquired, no per-doc loop)', function () {
+        this.ProjectLockManager.promises.extendLock.called.should.equal(false)
+      })
+
+      it('should drain the per-project queue', function () {
+        this.UpdateManager.promises.fetchAndApplyProjectUpdates.should.have.been.calledWith(
+          this.project_id
+        )
+      })
+    })
+  })
+
+  describe('fetchAndApplyDocUpdatesUnderDocLock', function () {
+    beforeEach(function () {
+      this.UpdateManager.promises.processOutstandingUpdates = sinon
+        .stub()
+        .resolves()
+    })
+
+    describe('when the doc has pending updates', function () {
+      beforeEach(async function () {
+        this.RealTimeRedisManager.promises.getUpdatesLength.resolves(2)
+        await this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock(
+          this.project_id,
+          this.doc_id,
+          new this.Profiler()
+        )
+      })
+
+      it('should take the per-doc lock', function () {
+        this.LockManager.promises.tryLock
+          .calledWith(this.doc_id)
           .should.equal(true)
       })
+
+      it('should process the doc updates and release the per-doc lock', function () {
+        this.UpdateManager.promises.processOutstandingUpdates.should.have.been.calledWith(
+          this.project_id,
+          this.doc_id
+        )
+        this.LockManager.promises.releaseLock
+          .calledWith(this.doc_id, this.lockValue)
+          .should.equal(true)
+      })
+    })
+
+    describe('when the doc has no pending updates', function () {
+      beforeEach(async function () {
+        this.RealTimeRedisManager.promises.getUpdatesLength.resolves(0)
+        await this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock(
+          this.project_id,
+          this.doc_id,
+          new this.Profiler()
+        )
+      })
+
+      it('should not take the per-doc lock', function () {
+        this.LockManager.promises.tryLock.called.should.equal(false)
+      })
+    })
+
+    describe('when the per-doc lock is taken (e.g. by an old instance)', function () {
+      beforeEach(async function () {
+        this.RealTimeRedisManager.promises.getUpdatesLength.resolves(2)
+        this.LockManager.promises.tryLock.resolves(null)
+        await this.UpdateManager.promises.fetchAndApplyDocUpdatesUnderDocLock(
+          this.project_id,
+          this.doc_id,
+          new this.Profiler()
+        )
+      })
+
+      it('should skip processing the doc', function () {
+        this.UpdateManager.promises.processOutstandingUpdates.called.should.equal(
+          false
+        )
+      })
+    })
+  })
+
+  describe('fetchAndApplyProjectUpdates', function () {
+    beforeEach(function () {
+      this.UpdateManager.promises.applyUpdate = sinon.stub().resolves()
+    })
+
+    describe('with updates carrying their doc id', function () {
+      beforeEach(async function () {
+        this.updates = [
+          { doc: 'doc-1', op: 'a' },
+          { doc: 'doc-2', op: 'b' },
+        ]
+        this.RealTimeRedisManager.promises.getPendingProjectUpdates.resolves(
+          this.updates
+        )
+        await this.UpdateManager.promises.fetchAndApplyProjectUpdates(
+          this.project_id,
+          new this.Profiler()
+        )
+      })
+
+      it('should read the per-project queue', function () {
+        this.RealTimeRedisManager.promises.getPendingProjectUpdates
+          .calledWith(this.project_id)
+          .should.equal(true)
+      })
+
+      it('should apply each update to the doc it carries', function () {
+        this.UpdateManager.promises.applyUpdate.should.have.been.calledWith(
+          this.project_id,
+          'doc-1',
+          this.updates[0]
+        )
+        this.UpdateManager.promises.applyUpdate.should.have.been.calledWith(
+          this.project_id,
+          'doc-2',
+          this.updates[1]
+        )
+      })
+    })
+
+    describe('with a history-OT update', function () {
+      beforeEach(async function () {
+        this.update = { doc: 'doc-1', op: 'a' }
+        this.RealTimeRedisManager.promises.getPendingProjectUpdates.resolves([
+          this.update,
+        ])
+        this.HistoryOTUpdateManager.isHistoryOTEditOperationUpdate.returns(true)
+        await this.UpdateManager.promises.fetchAndApplyProjectUpdates(
+          this.project_id,
+          new this.Profiler()
+        )
+      })
+
+      it('should route it to the history-OT update manager', function () {
+        this.HistoryOTUpdateManager.applyUpdate.should.have.been.calledWith(
+          this.project_id,
+          'doc-1',
+          this.update
+        )
+        this.UpdateManager.promises.applyUpdate.called.should.equal(false)
+      })
+    })
+
+    describe('with no updates', function () {
+      beforeEach(async function () {
+        this.RealTimeRedisManager.promises.getPendingProjectUpdates.resolves([])
+        await this.UpdateManager.promises.fetchAndApplyProjectUpdates(
+          this.project_id,
+          new this.Profiler()
+        )
+      })
+
+      it('should not apply any update', function () {
+        this.UpdateManager.promises.applyUpdate.called.should.equal(false)
+      })
     })
   })
 
   describe('continueProcessingUpdatesWithLock', function () {
     describe('when there are outstanding updates', function () {
       beforeEach(async function () {
-        this.RealTimeRedisManager.promises.getUpdatesLength.resolves(3)
+        this.UpdateManager.promises.getProjectPendingUpdatesLength = sinon
+          .stub()
+          .resolves(3)
         this.UpdateManager.promises.processOutstandingUpdatesWithLock = sinon
           .stub()
           .resolves()
@@ -282,7 +481,9 @@ describe('UpdateManager', function () {
 
     describe('when there are no outstanding updates', function () {
       beforeEach(async function () {
-        this.RealTimeRedisManager.promises.getUpdatesLength.resolves(0)
+        this.UpdateManager.promises.getProjectPendingUpdatesLength = sinon
+          .stub()
+          .resolves(0)
         this.UpdateManager.promises.processOutstandingUpdatesWithLock = sinon
           .stub()
           .resolves()
@@ -300,6 +501,61 @@ describe('UpdateManager', function () {
     })
   })
 
+  describe('getProjectPendingUpdatesLength', function () {
+    describe('when the per-project queue is non-empty', function () {
+      beforeEach(async function () {
+        this.RealTimeRedisManager.promises.getProjectUpdatesLength.resolves(4)
+        this.length =
+          await this.UpdateManager.promises.getProjectPendingUpdatesLength(
+            this.project_id,
+            this.doc_id
+          )
+      })
+
+      it('should short-circuit without scanning per-doc queues', function () {
+        this.length.should.equal(4)
+        this.RedisManager.promises.getDocIdsInProject.called.should.equal(false)
+      })
+    })
+
+    describe('when the per-project queue is empty (phases 1 & 2)', function () {
+      beforeEach(async function () {
+        this.Settings.pendingUpdatesMigrationPhase = 2
+        this.RealTimeRedisManager.promises.getProjectUpdatesLength.resolves(0)
+        this.RedisManager.promises.getDocIdsInProject.resolves(['doc-1'])
+        this.RealTimeRedisManager.promises.getUpdatesLength
+          .withArgs('doc-1')
+          .resolves(2)
+        this.length =
+          await this.UpdateManager.promises.getProjectPendingUpdatesLength(
+            this.project_id,
+            this.doc_id
+          )
+      })
+
+      it('should fall back to the per-doc queues', function () {
+        this.length.should.equal(2)
+      })
+    })
+
+    describe('in phase 3', function () {
+      beforeEach(async function () {
+        this.Settings.pendingUpdatesMigrationPhase = 3
+        this.RealTimeRedisManager.promises.getProjectUpdatesLength.resolves(0)
+        this.length =
+          await this.UpdateManager.promises.getProjectPendingUpdatesLength(
+            this.project_id,
+            this.doc_id
+          )
+      })
+
+      it('should only consider the per-project queue', function () {
+        this.length.should.equal(0)
+        this.RedisManager.promises.getDocIdsInProject.called.should.equal(false)
+      })
+    })
+  })
+
   describe('fetchAndApplyUpdates', function () {
     describe('with updates', function () {
       beforeEach(async function () {
@@ -902,7 +1158,7 @@ describe('UpdateManager', function () {
         this.UpdateManager.promises.continueProcessingUpdatesWithLock = sinon
           .stub()
           .resolves()
-        this.UpdateManager.promises.processOutstandingUpdates = sinon
+        this.UpdateManager.promises.processOutstandingProjectUpdates = sinon
           .stub()
           .resolves()
         this.response = await this.UpdateManager.promises.lockUpdatesAndDo(
@@ -925,10 +1181,11 @@ describe('UpdateManager', function () {
           .should.equal(true)
       })
 
-      it('should process any outstanding updates', function () {
-        this.UpdateManager.promises.processOutstandingUpdates.should.have.been.calledWith(
+      it('should process any outstanding updates for the project', function () {
+        this.UpdateManager.promises.processOutstandingProjectUpdates.should.have.been.calledWith(
           this.project_id,
-          this.doc_id
+          this.doc_id,
+          this.projectLockValue
         )
       })
 
@@ -958,10 +1215,10 @@ describe('UpdateManager', function () {
       })
     })
 
-    describe('when processOutstandingUpdates returns an error', function () {
+    describe('when processOutstandingProjectUpdates returns an error', function () {
       beforeEach(async function () {
         this.error = new Error('Something went wrong')
-        this.UpdateManager.promises.processOutstandingUpdates = sinon
+        this.UpdateManager.promises.processOutstandingProjectUpdates = sinon
           .stub()
           .rejects(this.error)
         await expect(
@@ -974,10 +1231,7 @@ describe('UpdateManager', function () {
         ).to.be.rejectedWith(this.error)
       })
 
-      it('should free both locks', function () {
-        this.LockManager.promises.releaseLock
-          .calledWith(this.doc_id, this.lockValue)
-          .should.equal(true)
+      it('should free the project lock', function () {
         this.ProjectLockManager.promises.releaseLock
           .calledWith(this.project_id, this.projectLockValue)
           .should.equal(true)
@@ -987,7 +1241,7 @@ describe('UpdateManager', function () {
     describe('when the method returns an error', function () {
       beforeEach(async function () {
         this.error = new Error('something went wrong')
-        this.UpdateManager.promises.processOutstandingUpdates = sinon
+        this.UpdateManager.promises.processOutstandingProjectUpdates = sinon
           .stub()
           .resolves()
         this.method = sinon.stub().rejects(this.error)