[monorepo] clean up per-project queue migration (#35573)

GitOrigin-RevId: 988df4cdc7d929efc921e7e99d803c6ce6e52c57
This commit is contained in:
Jakob Ackermann
2026-09-17 11:31:52 +00:00
committed by Copybot
parent d8f167dec2
commit 68352783b0
27 changed files with 195 additions and 1528 deletions
@@ -134,21 +134,15 @@ function _getPendingUpdateListKey() {
}
/**
* Queue an update for processing by the document-updater.
* Queue an update for processing by the document-updater: push the payload
* (tagged with its doc id) onto the project queue, then put a bare
* `project_id` marker onto pending-updates-list.
*
* @param {string} projectId
* @param {string} docId
* @param {Object} change
* @param {boolean} usePerProjectPendingUpdates - feature flag assigned on
* joinProject; in migration phase 2 it selects the per-project queue
* over the legacy per-doc queue
*/
async function queueChange(
projectId,
docId,
change,
usePerProjectPendingUpdates
) {
async function queueChange(projectId, docId, change) {
const allowedKeys = ['doc', 'op', 'v', 'dupIfSource', 'meta', 'lastV', 'hash']
change = _.pick(change, allowedKeys)
const jsonChange = JSON.stringify(change)
@@ -162,79 +156,14 @@ async function queueChange(
throw new UpdateTooLargeError(updateSize)
}
switch (settings.pendingUpdatesMigrationPhase) {
case 1:
await _queueChangeOnDocQueue(projectId, docId, jsonChange)
break
case 2:
// the consumer drains both queues in phase 2, so the feature flag can
// safely pick the queue per client
if (usePerProjectPendingUpdates) {
await _queueChangeOnProjectQueue(projectId, jsonChange)
} else {
await _queueChangeOnDocQueue(projectId, docId, jsonChange)
}
break
case 3:
await _queueChangeOnProjectQueue(projectId, jsonChange)
break
default:
throw new OError('invalid pendingUpdatesMigrationPhase', {
pendingUpdatesMigrationPhase: settings.pendingUpdatesMigrationPhase,
})
}
}
/**
* Queue an update on the legacy per-doc queue (phase 1): push the payload
* onto the doc queue, then put a `project_id:doc_id` marker onto
* pending-updates-list.
*
* @param {string} projectId
* @param {string} docId
* @param {string} jsonChange - the serialized change
*/
async function _queueChangeOnDocQueue(projectId, docId, jsonChange) {
// record metric for each update added to queue
metrics.summary('redis.pendingUpdates', jsonChange.length, {
status: 'push',
path: 'doc',
})
const docKey = `${projectId}:${docId}`
// Push onto pendingUpdates for doc_id first, because once the doc updater
// gets an entry on pending-updates-list, it starts processing.
try {
await rclient.rpush(Keys.pendingUpdates({ doc_id: docId }), jsonChange)
} catch (error) {
throw new OError('error pushing update into redis').withCause(error)
}
const queueKey = _getPendingUpdateListKey()
try {
await rclient.rpush(queueKey, docKey)
} catch (error) {
throw new OError('error pushing doc_id into redis')
.withInfo({ queueKey })
.withCause(error)
}
}
/**
* Queue an update on the per-project queue (phases 2 & 3): push the payload
* (tagged with its doc id) onto the project queue, then put a bare
* `project_id` marker onto pending-updates-list.
*
* @param {string} projectId
* @param {string} jsonChange
*/
async function _queueChangeOnProjectQueue(projectId, jsonChange) {
// record metric for each update added to queue
metrics.summary('redis.pendingUpdates', jsonChange.length, {
status: 'push',
path: 'project',
})
// Push onto the project queue first, because once the doc updater gets an
// entry on pending-updates-list, it starts processing.
try {
await rclient.rpush(
Keys.pendingProjectUpdates({ project_id: projectId }),
@@ -54,7 +54,6 @@ async function joinProject(projectId, user) {
isRestrictedUser: data.isRestrictedUser,
isTokenMember: data.isTokenMember,
isInvitedMember: data.isInvitedMember,
usePerProjectPendingUpdates: data.usePerProjectPendingUpdates,
}
return {
project: data.project,
@@ -101,8 +101,6 @@ export default WebsocketController = {
client.ol_context.is_restricted_user = !!userMetadata.isRestrictedUser
client.ol_context.is_token_member = !!userMetadata.isTokenMember
client.ol_context.is_invited_member = !!userMetadata.isInvitedMember
client.ol_context.use_per_project_pending_updates =
!!userMetadata.usePerProjectPendingUpdates
RoomManager.joinProject(client, projectId, function (err) {
if (err) {
@@ -560,11 +558,7 @@ export default WebsocketController = {
applyOtUpdate(client, docId, update, callback) {
// client may have disconnected, but we can submit their update to doc-updater anyways.
const {
user_id: userId,
project_id: projectId,
use_per_project_pending_updates: usePerProjectPendingUpdates,
} = client.ol_context
const { user_id: userId, project_id: projectId } = client.ol_context
if (!projectId) {
return callback(new NotJoinedError())
}
@@ -614,7 +608,6 @@ export default WebsocketController = {
projectId,
docId,
update,
usePerProjectPendingUpdates,
function (error) {
if ((error && error.message) === 'update is too large') {
metrics.inc('update_too_large')
@@ -61,9 +61,6 @@ const settings = {
process.env.REDIS_PASSWORD ||
'',
key_schema: {
pendingUpdates({ doc_id }) {
return `PendingUpdates:{${doc_id}}`
},
pendingProjectUpdates({ project_id }) {
return `PendingProjectUpdates:{${project_id}}`
},
@@ -134,20 +131,6 @@ const settings = {
10
),
// Migration of the document-updater update queue from a per-doc queue to a
// per-project queue. Must be kept in step with document-updater.
// phase 1: write the legacy per-doc queue (`PendingUpdates:{doc_id}` +
// `<project_id>:<doc_id>` marker)
// phase 2: write the per-project queue (`PendingProjectUpdates:{project_id}`
// + bare `<project_id>` marker) for clients with the
// per-project-pending-updates feature flag (assigned by web on
// joinProject), the legacy per-doc queue otherwise
// phase 3: write the per-project queue for all clients
pendingUpdatesMigrationPhase: parseInt(
process.env.PENDING_UPDATES_MIGRATION_PHASE || '1',
10
),
// combine
// max_doc_length (2mb see above) * 2 (delete + insert)
// max_ranges_size (3mb see MAX_RANGES_SIZE in document-updater)
@@ -107,11 +107,11 @@ describe('applyOtUpdate', function () {
)
})
it('should push the doc into the pending updates list', function (done) {
it('should push the project into the pending updates list', function (done) {
getPendingUpdatesList((error, ...rest) => {
if (error) return done(error)
const [docId] = Array.from(rest[0])
docId.should.equal(`${this.project_id}:${this.doc_id}`)
const [projectId] = Array.from(rest[0])
projectId.should.equal(this.project_id)
return done()
})
return null
@@ -119,8 +119,8 @@ describe('applyOtUpdate', function () {
it('should push the update into redis', function (done) {
rclient.lrange(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
0,
-1,
@@ -151,9 +151,9 @@ describe('applyOtUpdate', function () {
),
cb =>
rclient.del(
redisSettings.documentupdater.key_schema.pendingUpdates(
this.doc_id
),
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
cb
),
],
@@ -246,8 +246,8 @@ describe('applyOtUpdate', function () {
return it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -324,8 +324,8 @@ describe('applyOtUpdate', function () {
return it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -390,11 +390,11 @@ describe('applyOtUpdate', function () {
)
})
it('should push the doc into the pending updates list', function (done) {
it('should push the project into the pending updates list', function (done) {
getPendingUpdatesList((error, ...rest) => {
if (error) return done(error)
const [docId] = Array.from(rest[0])
docId.should.equal(`${this.project_id}:${this.doc_id}`)
const [projectId] = Array.from(rest[0])
projectId.should.equal(this.project_id)
return done()
})
return null
@@ -402,8 +402,8 @@ describe('applyOtUpdate', function () {
it('should push the update into redis', function (done) {
rclient.lrange(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
0,
-1,
@@ -434,8 +434,8 @@ describe('applyOtUpdate', function () {
),
cb =>
rclient.del(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
cb
),
@@ -503,8 +503,8 @@ describe('applyOtUpdate', function () {
cb => clearPendingUpdatesList(cb),
cb =>
rclient.del(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
cb
),
@@ -519,8 +519,8 @@ describe('applyOtUpdate', function () {
it('should put the update in redis', function (done) {
rclient.lrange(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
0,
-1,
@@ -601,8 +601,8 @@ describe('applyOtUpdate', function () {
return it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -675,8 +675,8 @@ describe('applyOtUpdate', function () {
it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -753,8 +753,8 @@ describe('applyOtUpdate', function () {
return it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -835,8 +835,8 @@ describe('applyOtUpdate', function () {
return it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -917,8 +917,8 @@ describe('applyOtUpdate', function () {
return it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -999,8 +999,8 @@ describe('applyOtUpdate', function () {
return it('should not put the update in redis', function (done) {
rclient.llen(
redisSettings.documentupdater.key_schema.pendingUpdates({
doc_id: this.doc_id,
redisSettings.documentupdater.key_schema.pendingProjectUpdates({
project_id: this.project_id,
}),
(error, len) => {
if (error) return done(error)
@@ -57,11 +57,16 @@ import redis from '@overleaf/redis-wrapper'
const Keys = settings.redis.documentupdater.key_schema
const rclient = redis.createClient(settings.redis.pubsub)
function getPendingUpdates(docId, cb) {
rclient.lrange(Keys.pendingUpdates({ doc_id: docId }), 0, 10, cb)
function getPendingUpdates(projectId, cb) {
rclient.lrange(
Keys.pendingProjectUpdates({ project_id: projectId }),
0,
10,
cb
)
}
function cleanupPreviousUpdates(docId, cb) {
rclient.del(Keys.pendingUpdates({ doc_id: docId }), cb)
function cleanupPreviousUpdates(projectId, cb) {
rclient.del(Keys.pendingProjectUpdates({ project_id: projectId }), cb)
}
describe('MatrixTests', function () {
@@ -411,7 +416,7 @@ describe('MatrixTests', function () {
let receivedArgs, submittedUpdates, update
beforeEach(function cleanup(done) {
cleanupPreviousUpdates(privateDocId, done)
cleanupPreviousUpdates(privateProjectId, done)
})
beforeEach(function setupUpdateFields() {
@@ -460,7 +465,7 @@ describe('MatrixTests', function () {
})
beforeEach(function fetchPendingOps(done) {
getPendingUpdates(privateDocId, (err, updates) => {
getPendingUpdates(privateProjectId, (err, updates) => {
submittedUpdates = updates
done(err)
})
@@ -74,8 +74,6 @@ export default MockWebServer = {
isRestrictedUser: !!userMetadata?.isRestrictedUser,
isTokenMember: !!userMetadata?.isTokenMember,
isInvitedMember: !!userMetadata?.isInvitedMember,
usePerProjectPendingUpdates:
!!userMetadata?.usePerProjectPendingUpdates,
})
}
)
@@ -22,11 +22,7 @@ describe('DocumentUpdaterController', function () {
default: (ctx.settings = {
redis: {
documentupdater: {
key_schema: {
pendingUpdates({ doc_id: docId }) {
return `PendingUpdates:${docId}`
},
},
key_schema: {},
},
pubsub: null,
},
@@ -21,9 +21,6 @@ describe('DocumentUpdaterManager', function () {
redis: {
documentupdater: {
key_schema: {
pendingUpdates({ doc_id: docId }) {
return `PendingUpdates:${docId}`
},
pendingProjectUpdates({ project_id: projectId }) {
return `PendingProjectUpdates:${projectId}`
},
@@ -32,7 +29,6 @@ describe('DocumentUpdaterManager', function () {
},
maxUpdateSize: 7 * 1024 * 1024,
pendingUpdateListShardCount: 10,
pendingUpdatesMigrationPhase: 1,
}
ctx.rclient = { auth() {} }
ctx.fetchJson = sinon.stub()
@@ -259,36 +255,6 @@ describe('DocumentUpdaterManager', function () {
)
})
it('should push the change', function (ctx) {
ctx.rclient.rpush
.calledWith(
`PendingUpdates:${ctx.doc_id}`,
JSON.stringify(ctx.change)
)
.should.equal(true)
})
it('should notify the doc updater of the change via the pending-updates-list queue', function (ctx) {
// The second call should be to a pending-updates-list key (either base or sharded)
const secondCall = ctx.rclient.rpush.secondCall
secondCall.should.exist
const queueKey = secondCall.args[0]
queueKey.should.match(/^pending-updates-list(-\d+)?$/)
secondCall.args[1].should.equal(`${ctx.project_id}:${ctx.doc_id}`)
})
})
describe('in migration phase 2 with the per-project queue flag', function () {
beforeEach(async function (ctx) {
ctx.settings.pendingUpdatesMigrationPhase = 2
await ctx.DocumentUpdaterManager.promises.queueChange(
ctx.project_id,
ctx.doc_id,
ctx.change,
true
)
})
it('should push the change onto the per-project queue with its doc id', function (ctx) {
ctx.rclient.rpush
.calledWith(
@@ -298,13 +264,8 @@ describe('DocumentUpdaterManager', function () {
.should.equal(true)
})
it('should not push onto the legacy per-doc queue', function (ctx) {
ctx.rclient.rpush
.calledWith(`PendingUpdates:${ctx.doc_id}`)
.should.equal(false)
})
it('should notify the doc updater with a bare project id marker', function (ctx) {
it('should notify the doc updater with a bare project id marker on the pending-updates-list queue', function (ctx) {
// The second call should be to a pending-updates-list key (either base or sharded)
const secondCall = ctx.rclient.rpush.secondCall
secondCall.should.exist
const queueKey = secondCall.args[0]
@@ -313,68 +274,6 @@ describe('DocumentUpdaterManager', function () {
})
})
describe('in migration phase 2 without the per-project queue flag', function () {
beforeEach(async function (ctx) {
ctx.settings.pendingUpdatesMigrationPhase = 2
await ctx.DocumentUpdaterManager.promises.queueChange(
ctx.project_id,
ctx.doc_id,
ctx.change,
false
)
})
it('should push the change onto the legacy per-doc queue', function (ctx) {
ctx.rclient.rpush
.calledWith(
`PendingUpdates:${ctx.doc_id}`,
JSON.stringify(ctx.change)
)
.should.equal(true)
})
it('should not push onto the per-project queue', function (ctx) {
ctx.rclient.rpush
.calledWith(`PendingProjectUpdates:${ctx.project_id}`)
.should.equal(false)
})
it('should notify the doc updater with a project_id:doc_id marker', function (ctx) {
const secondCall = ctx.rclient.rpush.secondCall
secondCall.should.exist
const queueKey = secondCall.args[0]
queueKey.should.match(/^pending-updates-list(-\d+)?$/)
secondCall.args[1].should.equal(`${ctx.project_id}:${ctx.doc_id}`)
})
})
describe('in migration phase 3 (per-project queue only)', function () {
beforeEach(async function (ctx) {
ctx.settings.pendingUpdatesMigrationPhase = 3
await ctx.DocumentUpdaterManager.promises.queueChange(
ctx.project_id,
ctx.doc_id,
ctx.change,
false
)
})
it('should push the change onto the per-project queue even without the flag', function (ctx) {
ctx.rclient.rpush
.calledWith(
`PendingProjectUpdates:${ctx.project_id}`,
JSON.stringify(ctx.change)
)
.should.equal(true)
})
it('should not push onto the legacy per-doc queue', function (ctx) {
ctx.rclient.rpush
.calledWith(`PendingUpdates:${ctx.doc_id}`)
.should.equal(false)
})
})
describe('with error talking to redis during rpush', function () {
beforeEach(function (ctx) {
ctx.rclient.rpush = sinon
@@ -482,7 +381,7 @@ describe('DocumentUpdaterManager', function () {
it('should remove the invalid keys from the change', function (ctx) {
ctx.rclient.rpush
.calledWith(
`PendingUpdates:${ctx.doc_id}`,
`PendingProjectUpdates:${ctx.project_id}`,
JSON.stringify({ op: ctx.change.op })
)
.should.equal(true)
@@ -42,7 +42,6 @@ describe('WebApiManager', function () {
isRestrictedUser: true,
isTokenMember: true,
isInvitedMember: true,
usePerProjectPendingUpdates: true,
}
ctx.fetchUtils.fetchJson.resolves(ctx.response)
ctx.WebApiManager.joinProject(ctx.project_id, ctx.user, ctx.callback)
@@ -71,8 +70,6 @@ describe('WebApiManager', function () {
isRestrictedUser: ctx.response.isRestrictedUser,
isTokenMember: ctx.response.isTokenMember,
isInvitedMember: ctx.response.isInvitedMember,
usePerProjectPendingUpdates:
ctx.response.usePerProjectPendingUpdates,
})
.should.equal(true)
})
@@ -92,7 +89,6 @@ describe('WebApiManager', function () {
isRestrictedUser: true,
isTokenMember: false,
isInvitedMember: false,
usePerProjectPendingUpdates: false,
}
ctx.fetchUtils.fetchJson.resolves(ctx.response)
ctx.WebApiManager.joinProject(ctx.project_id, ctx.user, ctx.callback)
@@ -124,8 +120,6 @@ describe('WebApiManager', function () {
isRestrictedUser: ctx.response.isRestrictedUser,
isTokenMember: ctx.response.isTokenMember,
isInvitedMember: ctx.response.isInvitedMember,
usePerProjectPendingUpdates:
ctx.response.usePerProjectPendingUpdates,
}
)
})
@@ -97,14 +97,12 @@ describe('WebsocketController', function () {
ctx.isRestrictedUser = true
ctx.isTokenMember = true
ctx.isInvitedMember = true
ctx.usePerProjectPendingUpdates = true
ctx.WebApiManager.joinProject = sinon
.stub()
.callsArgWith(2, null, ctx.project, ctx.privilegeLevel, {
isRestrictedUser: ctx.isRestrictedUser,
isTokenMember: ctx.isTokenMember,
isInvitedMember: ctx.isInvitedMember,
usePerProjectPendingUpdates: ctx.usePerProjectPendingUpdates,
})
ctx.RoomManager.joinProject = sinon.stub().callsArg(2)
return ctx.WebsocketController.joinProject(
@@ -178,11 +176,6 @@ describe('WebsocketController', function () {
ctx.isInvitedMember
)
})
it('should set the use_per_project_pending_updates flag on the client', function (ctx) {
ctx.client.ol_context.use_per_project_pending_updates.should.equal(
ctx.usePerProjectPendingUpdates
)
})
it('should call the callback with the project, privilegeLevel and protocolVersion', function (ctx) {
return ctx.callback
.calledWith(
@@ -247,14 +240,12 @@ describe('WebsocketController', function () {
ctx.isRestrictedUser = true
ctx.isTokenMember = true
ctx.isInvitedMember = true
ctx.usePerProjectPendingUpdates = true
ctx.WebApiManager.joinProject = sinon
.stub()
.callsArgWith(2, null, ctx.project, ctx.privilegeLevel, {
isRestrictedUser: ctx.isRestrictedUser,
isTokenMember: ctx.isTokenMember,
isInvitedMember: ctx.isInvitedMember,
usePerProjectPendingUpdates: ctx.usePerProjectPendingUpdates,
})
ctx.RoomManager.joinProject = sinon
.stub()
@@ -1475,11 +1466,10 @@ describe('WebsocketController', function () {
ctx.update = { op: { p: 12, t: 'foo' } }
ctx.client.ol_context.user_id = ctx.user_id
ctx.client.ol_context.project_id = ctx.project_id
ctx.client.ol_context.use_per_project_pending_updates = true
ctx.WebsocketController._assertClientCanApplyUpdate = sinon
.stub()
.yields()
ctx.DocumentUpdaterManager.queueChange = sinon.stub().callsArg(4)
ctx.DocumentUpdaterManager.queueChange = sinon.stub().callsArg(3)
})
describe('succesfully', function () {
@@ -1500,9 +1490,9 @@ describe('WebsocketController', function () {
ctx.update.meta.user_id.should.equal(ctx.user_id)
})
it('should queue the update with the per-project queue flag', function (ctx) {
it('should queue the update', function (ctx) {
ctx.DocumentUpdaterManager.queueChange
.calledWith(ctx.project_id, ctx.doc_id, ctx.update, true)
.calledWith(ctx.project_id, ctx.doc_id, ctx.update)
.should.equal(true)
})
@@ -1520,7 +1510,7 @@ describe('WebsocketController', function () {
ctx.client.disconnect = sinon.stub()
ctx.DocumentUpdaterManager.queueChange = sinon
.stub()
.callsArgWith(4, (ctx.error = new Error('Something went wrong')))
.callsArgWith(3, (ctx.error = new Error('Something went wrong')))
ctx.WebsocketController.applyOtUpdate(
ctx.client,
ctx.doc_id,
@@ -1580,7 +1570,7 @@ describe('WebsocketController', function () {
const error = new UpdateTooLargeError(7372835)
ctx.DocumentUpdaterManager.queueChange = sinon
.stub()
.callsArgWith(4, error)
.callsArgWith(3, error)
ctx.WebsocketController.applyOtUpdate(
ctx.client,
ctx.doc_id,