From 68352783b00d50fc7496ec31a6064086d174dead Mon Sep 17 00:00:00 2001 From: Jakob Ackermann Date: Mon, 20 Jul 2026 12:28:02 +0200 Subject: [PATCH] [monorepo] clean up per-project queue migration (#35573) GitOrigin-RevId: 988df4cdc7d929efc921e7e99d803c6ce6e52c57 --- .../app/js/DispatchManager.js | 11 +- .../app/js/RealTimeRedisManager.js | 36 +- .../document-updater/app/js/UpdateKeys.js | 10 +- .../document-updater/app/js/UpdateManager.js | 286 +-------------- .../config/settings.defaults.js | 14 - .../scripts/check_migration_phase_switch.js | 217 ----------- .../test/acceptance/js/QueueMigrationTests.js | 181 --------- .../acceptance/js/helpers/DocUpdaterClient.js | 41 +-- .../DispatchManager/DispatchManagerTests.js | 34 +- .../RealTimeRedisManagerTests.js | 82 ----- .../unit/js/RedisManager/RedisManagerTests.js | 3 - .../js/UpdateManager/UpdateManagerTests.js | 345 ++---------------- .../app/js/DocumentUpdaterManager.js | 83 +---- services/real-time/app/js/WebApiManager.js | 1 - .../real-time/app/js/WebsocketController.js | 9 +- .../real-time/config/settings.defaults.cjs | 17 - .../test/acceptance/js/ApplyUpdateTests.js | 70 ++-- .../test/acceptance/js/MatrixTests.js | 17 +- .../acceptance/js/helpers/MockWebServer.js | 2 - .../unit/js/DocumentUpdaterController.test.js | 6 +- .../unit/js/DocumentUpdaterManager.test.js | 107 +----- .../test/unit/js/WebApiManager.test.js | 6 - .../test/unit/js/WebsocketController.test.js | 20 +- .../Features/Editor/EditorHttpController.mjs | 11 - services/web/scripts/fix_oversized_docs.mjs | 45 ++- .../web/scripts/recover_docs_from_redis.mjs | 35 +- .../src/Editor/EditorHttpController.test.mjs | 34 -- 27 files changed, 195 insertions(+), 1528 deletions(-) delete mode 100644 services/document-updater/scripts/check_migration_phase_switch.js delete mode 100644 services/document-updater/test/acceptance/js/QueueMigrationTests.js diff --git a/services/document-updater/app/js/DispatchManager.js b/services/document-updater/app/js/DispatchManager.js index 95ac03b6f1..091ccf7631 100644 --- a/services/document-updater/app/js/DispatchManager.js +++ b/services/document-updater/app/js/DispatchManager.js @@ -15,7 +15,6 @@ let DispatchManager const Settings = require('@overleaf/settings') const logger = require('@overleaf/logger') -const Keys = require('./UpdateKeys') const redis = require('@overleaf/redis-wrapper') const Errors = require('./Errors') const _ = require('lodash') @@ -50,15 +49,11 @@ module.exports = DispatchManager = { if (result == null) { return callback() } - const [listName, docKey] = Array.from(result) - const [projectId, docId] = Array.from( - Keys.splitProjectIdAndDocId(docKey) - ) + const [listName, projectId] = Array.from(result) // Dispatch this in the background const backgroundTask = cb => UpdateManager.processOutstandingUpdatesWithLock( projectId, - docId, function (error) { // log everything except OpRangeNotAvailable errors, these are normal if (error != null) { @@ -68,12 +63,12 @@ module.exports = DispatchManager = { error instanceof Errors.DeleteMismatchError if (logAsDebug) { logger.debug( - { err: error, projectId, docId }, + { err: error, projectId }, 'error processing update' ) } else { logger.error( - { err: error, projectId, docId }, + { err: error, projectId }, 'error processing update' ) } diff --git a/services/document-updater/app/js/RealTimeRedisManager.js b/services/document-updater/app/js/RealTimeRedisManager.js index 80c8583c3a..1fac34c779 100644 --- a/services/document-updater/app/js/RealTimeRedisManager.js +++ b/services/document-updater/app/js/RealTimeRedisManager.js @@ -31,19 +31,16 @@ let COUNT = 0 const MAX_OPS_PER_ITERATION = 8 // process a limited number of ops for safety /** - * Shared implementation for draining a batch of updates from a - * pending-updates list. Used for both the per-doc and per-project queues; - * `path` ('doc' | 'project') differentiates the two in metrics. + * Drain a batch of updates from a pending-updates list. * * The MULTI only operates on the single passed key (which has an id in curly * braces), so all of its operations run on the same node in a cluster * environment. * * @param {string} key - the redis key of the queue - * @param {'doc' | 'project'} path - metrics label * @param {function(Error, Array=): void} callback */ -function getPendingUpdatesFromQueue(key, path, callback) { +function getPendingUpdatesFromQueue(key, callback) { const multi = rclient.multi() multi.llen(key) multi.lrange(key, 0, MAX_OPS_PER_ITERATION - 1) @@ -57,13 +54,13 @@ function getPendingUpdatesFromQueue(key, path, callback) { 'redis.pendingUpdates.llen', llen, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20, 25, 50, 75, 100], - { path } + { path: 'project' } ) for (const jsonUpdate of jsonUpdates) { // record metric for each update removed from queue metrics.summary('redis.pendingUpdates', jsonUpdate.length, { status: 'pop', - path, + path: 'project', }) } const updates = [] @@ -81,30 +78,6 @@ function getPendingUpdatesFromQueue(key, path, callback) { } const RealTimeRedisManager = { - /** - * Drain a batch of updates from a doc's legacy per-doc queue. - * - * @param {string} docId - * @param {function(Error, Array=): void} callback - */ - getPendingUpdatesForDoc(docId, callback) { - getPendingUpdatesFromQueue( - Keys.pendingUpdates({ doc_id: docId }), - 'doc', - callback - ) - }, - - /** - * Get the length of a doc's legacy per-doc queue. - * - * @param {string} docId - * @param {function(Error, number=): void} callback - */ - getUpdatesLength(docId, callback) { - rclient.llen(Keys.pendingUpdates({ doc_id: docId }), callback) - }, - /** * Drain a batch of updates from a project's per-project queue. Each update * on the per-project queue carries its own `doc` id so the caller knows @@ -116,7 +89,6 @@ const RealTimeRedisManager = { getPendingProjectUpdates(projectId, callback) { getPendingUpdatesFromQueue( Keys.pendingProjectUpdates({ project_id: projectId }), - 'project', callback ) }, diff --git a/services/document-updater/app/js/UpdateKeys.js b/services/document-updater/app/js/UpdateKeys.js index 374a109d25..7afd542011 100644 --- a/services/document-updater/app/js/UpdateKeys.js +++ b/services/document-updater/app/js/UpdateKeys.js @@ -2,7 +2,7 @@ // Fix any style issues and re-enable lint. module.exports = { /** - * Build a legacy `project_id:doc_id` dispatch marker. + * Build the `project_id:doc_id` key ShareJS uses to identify a doc. * * @param {string} projectId * @param {string} docId @@ -13,13 +13,11 @@ module.exports = { }, /** - * Split a dispatch marker into project id and doc id. A marker is either - * the legacy `project_id:doc_id` or, after the queue migration, a bare - * `project_id`. Project and doc ids never contain a colon, so splitting is - * unambiguous; for the bare form docId is undefined. + * Split a `project_id:doc_id` ShareJS doc key into project id and doc id. + * Project and doc ids never contain a colon, so splitting is unambiguous. * * @param {string} projectAndDocId - * @return {[string, string | undefined]} + * @return {[string, string]} */ splitProjectIdAndDocId(projectAndDocId) { const [projectId, docId] = projectAndDocId.split(':') diff --git a/services/document-updater/app/js/UpdateManager.js b/services/document-updater/app/js/UpdateManager.js index b02a19d688..e8cf51da6e 100644 --- a/services/document-updater/app/js/UpdateManager.js +++ b/services/document-updater/app/js/UpdateManager.js @@ -1,6 +1,5 @@ // @ts-check -const Settings = require('@overleaf/settings') const { callbackifyAll } = require('@overleaf/promise-utils') const LockManager = require('./LockManager') const ProjectLockManager = require('./ProjectLockManager') @@ -25,35 +24,15 @@ const HistoryOTUpdateManager = require('./HistoryOTUpdateManager') */ const UpdateManager = { - /** - * Process the pending updates on a single doc's legacy per-doc queue. The - * caller must hold the doc lock. - * - * @param {string} projectId - * @param {string} docId - */ - async processOutstandingUpdates(projectId, docId) { - const timer = new Metrics.Timer('updateManager.processOutstandingUpdates') - try { - await UpdateManager.fetchAndApplyUpdates(projectId, docId) - timer.done({ status: 'success' }) - } catch (err) { - timer.done({ status: 'error' }) - throw err - } - }, - /** * Process the pending updates for a project under the project lock, then * keep going for as long as more updates are queued up. * * @param {string} projectId - * @param {string} [docIdHint] - doc id from a legacy dispatch marker, if any */ - async processOutstandingUpdatesWithLock(projectId, docIdHint) { + async processOutstandingUpdatesWithLock(projectId) { const profile = new Profiler('processOutstandingUpdatesWithLock', { project_id: projectId, - doc_id: docIdHint, }) // Always take the project lock before the per-doc lock to avoid deadlocks. @@ -64,131 +43,14 @@ const UpdateManager = { profile.log('getProjectLock') try { - await UpdateManager.processOutstandingProjectUpdates( - projectId, - docIdHint, - projectLockValue, - profile - ) - profile.log('processOutstandingProjectUpdates') + await UpdateManager.fetchAndApplyProjectUpdates(projectId, profile) + profile.log('fetchAndApplyProjectUpdates') } finally { await ProjectLockManager.promises.releaseLock(projectId, projectLockValue) profile.log('releaseProjectLock').end() } - await UpdateManager.continueProcessingUpdatesWithLock(projectId, docIdHint) - }, - - /** - * Process the pending updates for a project while holding the project lock. - * - * In phases 1 & 2 this drains the legacy per-doc queues (taking the per-doc - * lock for each doc) as well as the shared per-project queue; in phase 3 - * only the per-project queue is processed. - * - * @param {string} projectId - * @param {string | undefined} docIdHint - doc id from a legacy dispatch - * marker, if any - * @param {string} projectLockValue - lock value of the held project lock - * @param {Profiler} profile - profile started by the caller - */ - async processOutstandingProjectUpdates( - projectId, - docIdHint, - projectLockValue, - profile - ) { - switch (Settings.pendingUpdatesMigrationPhase) { - case 1: - case 2: - await UpdateManager.processOutstandingDocUpdates( - projectId, - docIdHint, - projectLockValue, - profile - ) - // Phases 1 & 2 drain the per-project queue as well, as a safety net in - // case a producer writes it ahead of the phase switch. - // falls through - case 3: - await UpdateManager.fetchAndApplyProjectUpdates(projectId, profile) - break - default: - throw new Error( - `invalid pendingUpdatesMigrationPhase: ${Settings.pendingUpdatesMigrationPhase}` - ) - } - }, - - /** - * Drain the legacy per-doc queues of all the docs in the project, plus the - * hinted doc, while holding the project lock. - * - * @param {string} projectId - * @param {string | undefined} docIdHint - doc id from a legacy dispatch - * marker, if any - * @param {string} projectLockValue - lock value of the held project lock - * @param {Profiler} profile - profile started by the caller - */ - async processOutstandingDocUpdates( - projectId, - docIdHint, - projectLockValue, - profile - ) { - const docIds = new Set( - await RedisManager.promises.getDocIdsInProject(projectId) - ) - profile.log('getDocIdsInProject') - if (docIdHint) { - docIds.add(docIdHint) - } - // Listing the docs above may have taken a while, so bump before we start - // processing and again after each doc. The final bump also covers draining - // the per-project queue afterwards. - await ProjectLockManager.promises.extendLock(projectId, projectLockValue) - profile.log('extendProjectLock') - for (const docId of docIds) { - await UpdateManager.fetchAndApplyDocUpdatesUnderDocLock( - projectId, - docId, - profile - ) - await ProjectLockManager.promises.extendLock(projectId, projectLockValue) - profile.log('extendProjectLock') - } - }, - - /** - * Drain (one batch of) a single doc's legacy per-doc queue, taking the - * per-doc lock so that an old document-updater instance - which only takes - * the per-doc lock - cannot process the same doc concurrently during a - * rolling deploy. tryLock and skip on contention; the doc is revisited by - * the continuation check or a later dispatch. - * - * @param {string} projectId - * @param {string} docId - * @param {Profiler} profile - profile started by the caller - */ - async fetchAndApplyDocUpdatesUnderDocLock(projectId, docId, profile) { - const length = await RealTimeRedisManager.promises.getUpdatesLength(docId) - profile.log('getUpdatesLength') - if (length === 0) { - return - } - const lockValue = await LockManager.promises.tryLock(docId) - if (lockValue == null) { - return - } - profile.log('tryLock') - - try { - await UpdateManager.processOutstandingUpdates(projectId, docId) - profile.log('processOutstandingUpdates') - } finally { - await LockManager.promises.releaseLock(docId, lockValue) - profile.log('releaseLock') - } + await UpdateManager.continueProcessingUpdatesWithLock(projectId) }, /** @@ -218,121 +80,20 @@ const UpdateManager = { } }, - /** - * Total pending updates for a project: the per-project queue plus (in phases - * 1 & 2) the loaded docs' legacy per-doc queues. Used to decide whether to - * keep processing after releasing the lock - other workers' markers may have - * been consumed (and dropped on a failed tryLock) while we held it. - * - * @param {string} projectId - * @param {string} [docIdHint] - doc id from a legacy dispatch marker, if any - * @return {Promise} - */ - async getProjectPendingUpdatesLength(projectId, docIdHint) { - switch (Settings.pendingUpdatesMigrationPhase) { - case 1: - case 2: { - const length = - await RealTimeRedisManager.promises.getProjectUpdatesLength(projectId) - if (length > 0) { - return length - } - return await UpdateManager.getDocPendingUpdatesLength( - projectId, - docIdHint - ) - } - case 3: - return await RealTimeRedisManager.promises.getProjectUpdatesLength( - projectId - ) - default: - throw new Error( - `invalid pendingUpdatesMigrationPhase: ${Settings.pendingUpdatesMigrationPhase}` - ) - } - }, - - /** - * Pending updates across the legacy per-doc queues of the docs in the - * project, plus the hinted doc. Returns early with the first non-empty - * queue's length. - * - * @param {string} projectId - * @param {string} [docIdHint] - doc id from a legacy dispatch marker, if any - * @return {Promise} - */ - async getDocPendingUpdatesLength(projectId, docIdHint) { - const docIds = new Set( - await RedisManager.promises.getDocIdsInProject(projectId) - ) - if (docIdHint) { - docIds.add(docIdHint) - } - for (const docId of docIds) { - const length = await RealTimeRedisManager.promises.getUpdatesLength(docId) - if (length > 0) { - return length - } - } - return 0 - }, - /** * Process the project's pending updates under the project lock if any are * queued up. * * @param {string} projectId - * @param {string} [docIdHint] - doc id from a legacy dispatch marker, if any */ - async continueProcessingUpdatesWithLock(projectId, docIdHint) { - const length = await UpdateManager.getProjectPendingUpdatesLength( - projectId, - docIdHint - ) + async continueProcessingUpdatesWithLock(projectId) { + const length = + await RealTimeRedisManager.promises.getProjectUpdatesLength(projectId) if (length > 0) { - await UpdateManager.processOutstandingUpdatesWithLock( - projectId, - docIdHint - ) + await UpdateManager.processOutstandingUpdatesWithLock(projectId) } }, - /** - * Apply one batch of updates from a doc's legacy per-doc queue. The caller - * must hold the doc lock. - * - * @param {string} projectId - * @param {string} docId - */ - async fetchAndApplyUpdates(projectId, docId) { - const profile = new Profiler('fetchAndApplyUpdates', { - project_id: projectId, - doc_id: docId, - }) - - const updates = - await RealTimeRedisManager.promises.getPendingUpdatesForDoc(docId) - logger.debug( - { projectId, docId, count: updates.length }, - 'processing updates' - ) - if (updates.length === 0) { - return - } - profile.log('getPendingUpdatesForDoc') - - for (const update of updates) { - if (HistoryOTUpdateManager.isHistoryOTEditOperationUpdate(update)) { - await HistoryOTUpdateManager.applyUpdate(projectId, docId, update) - } else { - await UpdateManager.applyUpdate(projectId, docId, update) - } - profile.log('applyUpdate') - } - profile.log('async done').end() - }, - /** * Apply an update to the given document * @@ -539,13 +300,8 @@ const UpdateManager = { let result try { - await UpdateManager.processOutstandingProjectUpdates( - projectId, - docId, - projectLockValue, - profile - ) - profile.log('processOutstandingProjectUpdates') + await UpdateManager.fetchAndApplyProjectUpdates(projectId, profile) + profile.log('fetchAndApplyProjectUpdates') const lockValue = await LockManager.promises.getLock(docId) profile.log('getLock') @@ -563,18 +319,16 @@ const UpdateManager = { } // We held the lock for a while so updates might have queued up - UpdateManager.continueProcessingUpdatesWithLock(projectId, docId).catch( - err => { - // The processing may fail for invalid user updates. - // This can be very noisy, put them on level DEBUG - // and record a metric. - Metrics.inc('background-processing-updates-error') - logger.debug( - { err, projectId, docId }, - 'error processing updates in background' - ) - } - ) + UpdateManager.continueProcessingUpdatesWithLock(projectId).catch(err => { + // The processing may fail for invalid user updates. + // This can be very noisy, put them on level DEBUG + // and record a metric. + Metrics.inc('background-processing-updates-error') + logger.debug( + { err, projectId, docId }, + 'error processing updates in background' + ) + }) return result }, diff --git a/services/document-updater/config/settings.defaults.js b/services/document-updater/config/settings.defaults.js index d0e1db8fe5..6645c72c8a 100755 --- a/services/document-updater/config/settings.defaults.js +++ b/services/document-updater/config/settings.defaults.js @@ -140,9 +140,6 @@ module.exports = { projectBlock({ project_id: projectId }) { return `ProjectBlock:{${projectId}}` }, - pendingUpdates({ doc_id: docId }) { - return `PendingUpdates:{${docId}}` - }, pendingProjectUpdates({ project_id: projectId }) { return `PendingProjectUpdates:{${projectId}}` }, @@ -174,17 +171,6 @@ module.exports = { dispatcherCount: parseInt(process.env.DISPATCHER_COUNT || 10, 10), - // Migration of the real-time -> document-updater update queue from a per-doc - // queue to a per-project queue. See - // services/document-updater/scripts/check_migration_phase_switch.js - // phase 1: legacy per-doc queue only - // phase 2: producer writes the per-project queue; consumer drains both - // phase 3: per-project queue only - pendingUpdatesMigrationPhase: parseInt( - process.env.PENDING_UPDATES_MIGRATION_PHASE || '1', - 10 - ), - redisLockTTLSeconds: 30, mongo: { diff --git a/services/document-updater/scripts/check_migration_phase_switch.js b/services/document-updater/scripts/check_migration_phase_switch.js deleted file mode 100644 index ad9e87573f..0000000000 --- a/services/document-updater/scripts/check_migration_phase_switch.js +++ /dev/null @@ -1,217 +0,0 @@ -/** - * Pre-flight check for switching the real-time -> document-updater update queue - * migration phase (PENDING_UPDATES_MIGRATION_PHASE). - * - * phase 1: legacy per-doc queue only - * phase 2: producer writes the per-project queue; consumer drains both - * phase 3: per-project queue only - * - * Going to phase 3 is only safe once no data can be stranded in the legacy - * per-doc queues (because the phase-3 consumer stops reading them). This script - * verifies that and exits non-zero when the switch is unsafe. - * - * It is READ-ONLY (SCAN / LLEN / LRANGE only - never KEYS) and safe to run at - * any time. In a redis cluster it SCANs every master node, since the per-doc - * keys are spread across slots. - * - * Usage: - * node scripts/check_migration_phase_switch.js --current 2 --target 3 - */ -const { setTimeout } = require('node:timers/promises') -const minimist = require('minimist') -const Settings = require('@overleaf/settings') -const { rclient } = require('./../app/js/RedisManager') -const logger = require('@overleaf/logger') - -// Window size when walking a dispatch list tail -> head. -const LIST_CHUNK_SIZE = 1000 - -/** - * List all cluster master nodes, or the plain client outside a cluster. - * - * @return {Array} - */ -function getNodes() { - return ( - (typeof rclient.nodes === 'function' - ? rclient.nodes('master') - : undefined) || [rclient] - ) -} - -/** - * Non-blocking SCAN across all cluster master nodes. - * - * @param {string} pattern - a redis MATCH pattern - * @yields {string} matching keys - */ -async function* scanKeys(pattern) { - for (const node of getNodes()) { - let cursor = '0' - do { - const [next, keys] = await node.scan( - cursor, - 'MATCH', - pattern, - 'COUNT', - 1000 - ) - cursor = next - for (const key of keys) { - yield key - } - if (cursor !== '0') { - await setTimeout(10) // be gentle on redis - } - } while (cursor !== '0') - } -} - -/** - * Are there any non-empty legacy per-doc queues left? - * - * @return {Promise<{nonEmptyDocs: number, totalOps: number}>} - */ -async function checkPerDocQueues() { - let nonEmptyDocs = 0 - let totalOps = 0 - for await (const key of scanKeys('PendingUpdates:{*}')) { - const length = await rclient.llen(key) - logger.info({ key, length }, 'found legacy per-doc queue') - if (length > 0) { - nonEmptyDocs++ - totalOps += length - } - } - return { nonEmptyDocs, totalOps } -} - -/** - * Are there any legacy ":" markers still sitting on the dispatch - * lists? A colon means a phase-1 producer is still writing. We snapshot the - * length and walk each list tail -> head in bounded windows, rather than - * materialising the whole list in one LRANGE. The lists are only popped from - * the head (BLPOP) and appended at the tail (RPUSH), so an entry's index only - * ever decreases; sweeping tail -> head with contiguous windows therefore - * never skips an entry that is still queued (though a fast-draining entry can - * ride the window boundary down and be read several times - harmless for a - * colon scan). The only entries we don't observe are ones drained mid-walk - * (already processed) or pushed after we started. The exhaustive per-doc SCAN - * above is in any case the authoritative gate; a colon found here is just a - * strong extra signal. - * - * @return {Promise<{legacyMarkers: number, inspectedEntries: number}>} - */ -async function checkDispatchListsForLegacyMarkers() { - const shardCount = Settings.dispatcherCount - let legacyMarkers = 0 - let inspectedEntries = 0 - for (let shard = 0; shard < shardCount; shard++) { - const listKey = - shard === 0 ? 'pending-updates-list' : `pending-updates-list-${shard}` - const length = await rclient.llen(listKey) - for (let end = length; end > 0; end -= LIST_CHUNK_SIZE) { - const start = Math.max(0, end - LIST_CHUNK_SIZE) - const entries = await rclient.lrange(listKey, start, end - 1) - for (const marker of entries) { - inspectedEntries++ - if (marker.includes(':')) { - logger.info({ marker }, 'found legacy per-doc queue marker') - legacyMarkers++ - } - } - } - } - return { legacyMarkers, inspectedEntries } -} - -/** - * Check whether switching the migration phase is safe. - * - * @param {number} current - the phase currently running - * @param {number} target - the phase to switch to - * @return {Promise} whether the switch is safe - */ -async function checkSwitch(current, target) { - console.log(`Checking switch from phase ${current} -> ${target}`) - - if (Math.abs(target - current) > 1) { - console.warn( - `WARNING: ${current} -> ${target} is not an adjacent transition. ` + - 'Roll the phase one step at a time.' - ) - } - - // The only gated transition is enabling phase 3 (consumer stops reading the - // per-doc queues). Everything else is safe with the dual-read consumer. - if (target < 3) { - console.log( - 'Target phase < 3: the dual-read consumer drains both queues, so this ' + - 'switch (incl. roll-backs) is safe.' - ) - return true - } - - const { nonEmptyDocs, totalOps } = await checkPerDocQueues() - const { legacyMarkers, inspectedEntries } = - await checkDispatchListsForLegacyMarkers() - - console.log( - `Legacy per-doc queues with pending updates: ${nonEmptyDocs} (${totalOps} ops)` - ) - console.log( - `Legacy ":" markers found across ${inspectedEntries} dispatch ` + - `list entries: ${legacyMarkers}` - ) - - const safe = nonEmptyDocs === 0 && legacyMarkers === 0 - if (safe) { - console.log( - 'SAFE: no legacy per-doc data remains; phase 3 will not strand updates.' - ) - } else { - console.error( - 'UNSAFE: legacy per-doc data still present. Keep the producer at phase ' + - '>= 2 and wait for the per-doc queues to drain before switching to ' + - 'phase 3.' - ) - } - return safe -} - -/** - * Parse the command line arguments and run the check. - * - * @return {Promise} whether the switch is safe - */ -async function main() { - const argv = minimist(process.argv.slice(2), { - boolean: ['help'], - alias: { h: 'help' }, - }) - - if (argv.help || argv.current == null || argv.target == null) { - console.log(` -Usage: node scripts/check_migration_phase_switch.js --current N --target M - -Options: - --current The phase currently running (1, 2 or 3) - --target The phase you want to switch to (1, 2 or 3) - --help, -h Show this help message -`) - process.exit(argv.help ? 0 : 1) - } - - const current = parseInt(argv.current, 10) - const target = parseInt(argv.target, 10) - return await checkSwitch(current, target) -} - -main() - .then(safe => { - process.exit(safe ? 0 : 1) - }) - .catch(error => { - console.error('Error running phase-switch check', error) - process.exit(2) - }) diff --git a/services/document-updater/test/acceptance/js/QueueMigrationTests.js b/services/document-updater/test/acceptance/js/QueueMigrationTests.js deleted file mode 100644 index 8b76c3dfaa..0000000000 --- a/services/document-updater/test/acceptance/js/QueueMigrationTests.js +++ /dev/null @@ -1,181 +0,0 @@ -/* - * 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) - }) - }) -}) diff --git a/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js b/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js index d05cab3c3d..4646c11022 100644 --- a/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js +++ b/services/document-updater/test/acceptance/js/helpers/DocUpdaterClient.js @@ -36,52 +36,35 @@ module.exports = DocUpdaterClient = { rclientSub.on('message', messageHandler) }, - // 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 + ":" marker - // phase >= 2 / toProjectQueue=true: per-project queue + bare "" 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}`) - } + // Enqueue an update for the doc-updater: push the payload (tagged with its + // doc id) onto the per-project queue, then a bare "" marker onto the + // dispatch list. + async sendUpdate(projectId, docId, update) { + await rclient.rpush( + keys.pendingProjectUpdates({ project_id: projectId }), + JSON.stringify({ ...update, doc: docId }) + ) + await rclient.rpush(getPendingUpdateListKey(), projectId) }, - async sendUpdates(projectId, docId, updates, options = {}) { + async sendUpdates(projectId, docId, updates) { await DocUpdaterClient.preloadDoc(projectId, docId) for (const update of updates) { - await DocUpdaterClient.sendUpdate(projectId, docId, update, options) + await DocUpdaterClient.sendUpdate(projectId, docId, update) } await DocUpdaterClient.waitForPendingUpdates(projectId, 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 docLength = await rclient.llen( - keys.pendingUpdates({ doc_id: docId }) - ) const projectLength = await rclient.llen( keys.pendingProjectUpdates({ project_id: projectId }) ) - if (docLength === 0 && projectLength === 0) { + if (projectLength === 0) { return // Success - no pending updates } diff --git a/services/document-updater/test/unit/js/DispatchManager/DispatchManagerTests.js b/services/document-updater/test/unit/js/DispatchManager/DispatchManagerTests.js index 84479b3c05..845f03546f 100644 --- a/services/document-updater/test/unit/js/DispatchManager/DispatchManagerTests.js +++ b/services/document-updater/test/unit/js/DispatchManager/DispatchManagerTests.js @@ -68,18 +68,16 @@ describe('DispatchManager', function () { describe('_waitForUpdateThenDispatchWorker', function () { beforeEach(function () { this.project_id = 'project-id-123' - this.doc_id = 'doc-id-123' - this.doc_key = `${this.project_id}:${this.doc_id}` return (this.client.blpop = sinon .stub() - .callsArgWith(2, null, ['pending-updates-list', this.doc_key])) + .callsArgWith(2, null, ['pending-updates-list', this.project_id])) }) describe('in the normal case', function () { beforeEach(function () { this.UpdateManager.processOutstandingUpdatesWithLock = sinon .stub() - .callsArg(2) + .callsArg(1) return this.worker._waitForUpdateThenDispatchWorker(this.callback) }) @@ -91,7 +89,7 @@ describe('DispatchManager', function () { it('should call processOutstandingUpdatesWithLock', function () { return this.UpdateManager.processOutstandingUpdatesWithLock - .calledWith(this.project_id, this.doc_id) + .calledWith(this.project_id) .should.equal(true) }) @@ -109,7 +107,7 @@ describe('DispatchManager', function () { beforeEach(function () { this.UpdateManager.processOutstandingUpdatesWithLock = sinon .stub() - .callsArgWith(2, new Error('a generic error')) + .callsArgWith(1, new Error('a generic error')) return this.worker._waitForUpdateThenDispatchWorker(this.callback) }) @@ -126,7 +124,7 @@ describe('DispatchManager', function () { beforeEach(function () { this.UpdateManager.processOutstandingUpdatesWithLock = sinon .stub() - .callsArgWith(2, new Errors.DeleteMismatchError()) + .callsArgWith(1, new Errors.DeleteMismatchError()) return this.worker._waitForUpdateThenDispatchWorker(this.callback) }) @@ -139,28 +137,6 @@ 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 = { diff --git a/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js b/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js index 0604cbf9ca..99c3b7fe1a 100644 --- a/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js +++ b/services/document-updater/test/unit/js/RealTimeRedisManager/RealTimeRedisManagerTests.js @@ -32,9 +32,6 @@ describe('RealTimeRedisManager', function () { redis: { documentupdater: (this.settings = { key_schema: { - pendingUpdates({ doc_id: docId }) { - return `PendingUpdates:${docId}` - }, pendingProjectUpdates({ project_id: projectId }) { return `PendingProjectUpdates:${projectId}` }, @@ -64,85 +61,6 @@ describe('RealTimeRedisManager', function () { return (this.callback = sinon.stub()) }) - describe('getPendingUpdatesForDoc', function () { - beforeEach(function () { - this.rclient.llen = sinon.stub() - this.rclient.lrange = sinon.stub() - return (this.rclient.ltrim = sinon.stub()) - }) - - describe('successfully', function () { - beforeEach(function () { - this.updates = [ - { op: [{ i: 'foo', p: 4 }] }, - { op: [{ i: 'foo', p: 4 }] }, - ] - this.jsonUpdates = this.updates.map(update => JSON.stringify(update)) - this.rclient.exec = sinon.stub().yields(null, [2, this.jsonUpdates]) - return this.RealTimeRedisManager.getPendingUpdatesForDoc( - this.doc_id, - this.callback - ) - }) - - it('should get the pending updates', function () { - return this.rclient.lrange - .calledWith(`PendingUpdates:${this.doc_id}`, 0, 7) - .should.equal(true) - }) - - it('should delete the pending updates', function () { - return this.rclient.ltrim - .calledWith(`PendingUpdates:${this.doc_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({ op: [{ i: 'foo', p: 4 }] }), - 'broken json', - ] - this.rclient.exec = sinon.stub().yields(null, [2, this.jsonUpdates]) - return this.RealTimeRedisManager.getPendingUpdatesForDoc( - this.doc_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('getUpdatesLength', function () { - beforeEach(function () { - this.rclient.llen = sinon.stub().yields(null, (this.length = 3)) - return this.RealTimeRedisManager.getUpdatesLength( - this.doc_id, - this.callback - ) - }) - - it('should look up the length', function () { - return this.rclient.llen - .calledWith(`PendingUpdates:${this.doc_id}`) - .should.equal(true) - }) - - return it('should return the length', function () { - return this.callback.calledWith(null, this.length).should.equal(true) - }) - }) - describe('getPendingProjectUpdates', function () { beforeEach(function () { this.rclient.llen = sinon.stub() diff --git a/services/document-updater/test/unit/js/RedisManager/RedisManagerTests.js b/services/document-updater/test/unit/js/RedisManager/RedisManagerTests.js index 7e3919e6f8..faa18df8d7 100644 --- a/services/document-updater/test/unit/js/RedisManager/RedisManagerTests.js +++ b/services/document-updater/test/unit/js/RedisManager/RedisManagerTests.js @@ -37,9 +37,6 @@ describe('RedisManager', function () { projectKey({ doc_id: docId }) { return `ProjectId:${docId}` }, - pendingUpdates({ doc_id: docId }) { - return `PendingUpdates:${docId}` - }, docsInProject({ project_id: projectId }) { return `DocsIn:${projectId}` }, diff --git a/services/document-updater/test/unit/js/UpdateManager/UpdateManagerTests.js b/services/document-updater/test/unit/js/UpdateManager/UpdateManagerTests.js index f33611316a..4e8b4ab521 100644 --- a/services/document-updater/test/unit/js/UpdateManager/UpdateManagerTests.js +++ b/services/document-updater/test/unit/js/UpdateManager/UpdateManagerTests.js @@ -26,7 +26,6 @@ describe('UpdateManager', function () { this.LockManager = { promises: { - tryLock: sinon.stub().resolves(this.lockValue), getLock: sinon.stub().resolves(this.lockValue), releaseLock: sinon.stub().resolves(), }, @@ -34,9 +33,7 @@ describe('UpdateManager', function () { this.ProjectLockManager = { promises: { - tryLock: sinon.stub().resolves(this.projectLockValue), getLock: sinon.stub().resolves(this.projectLockValue), - extendLock: sinon.stub().resolves(), releaseLock: sinon.stub().resolves(), }, } @@ -46,15 +43,12 @@ 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().resolves(0), - getPendingUpdatesForDoc: sinon.stub(), getProjectUpdatesLength: sinon.stub().resolves(0), getPendingProjectUpdates: sinon.stub().resolves([]), }, @@ -75,8 +69,6 @@ describe('UpdateManager', function () { recordAndFlushHistoryOps: sinon.stub(), } - this.Settings = { pendingUpdatesMigrationPhase: 1 } - this.DocumentManager = { promises: { getDoc: sinon.stub(), @@ -117,7 +109,6 @@ describe('UpdateManager', function () { './HistoryManager': this.HistoryManager, './HistoryOTUpdateManager': this.HistoryOTUpdateManager, './Metrics': this.Metrics, - '@overleaf/settings': this.Settings, './DocumentManager': this.DocumentManager, './RangesManager': this.RangesManager, './SnapshotManager': this.SnapshotManager, @@ -128,33 +119,13 @@ describe('UpdateManager', function () { }) }) - describe('processOutstandingUpdates', function () { - beforeEach(async function () { - this.UpdateManager.promises.fetchAndApplyUpdates = sinon.stub().resolves() - await this.UpdateManager.promises.processOutstandingUpdates( - this.project_id, - this.doc_id - ) - }) - - it('should apply the updates', function () { - this.UpdateManager.promises.fetchAndApplyUpdates - .calledWith(this.project_id, this.doc_id) - .should.equal(true) - }) - - it('should time the execution', function () { - this.Metrics.Timer.prototype.done.called.should.equal(true) - }) - }) - describe('processOutstandingUpdatesWithLock', function () { describe('when the project lock is free', function () { beforeEach(function () { this.UpdateManager.promises.continueProcessingUpdatesWithLock = sinon .stub() .resolves() - this.UpdateManager.promises.processOutstandingProjectUpdates = sinon + this.UpdateManager.promises.fetchAndApplyProjectUpdates = sinon .stub() .resolves() }) @@ -162,8 +133,7 @@ describe('UpdateManager', function () { describe('successfully', function () { beforeEach(async function () { await this.UpdateManager.promises.processOutstandingUpdatesWithLock( - this.project_id, - this.doc_id + this.project_id ) }) @@ -179,38 +149,37 @@ describe('UpdateManager', function () { .should.equal(true) }) - it('should process the outstanding project updates', function () { - this.UpdateManager.promises.processOutstandingProjectUpdates - .calledWith(this.project_id, this.doc_id, this.projectLockValue) + it('should drain the per-project queue', function () { + this.UpdateManager.promises.fetchAndApplyProjectUpdates + .calledWith(this.project_id) .should.equal(true) }) it('should do everything with the lock acquired', function () { - this.UpdateManager.promises.processOutstandingProjectUpdates + this.UpdateManager.promises.fetchAndApplyProjectUpdates .calledAfter(this.ProjectLockManager.promises.getLock) .should.equal(true) - this.UpdateManager.promises.processOutstandingProjectUpdates + this.UpdateManager.promises.fetchAndApplyProjectUpdates .calledBefore(this.ProjectLockManager.promises.releaseLock) .should.equal(true) }) it('should continue processing new updates that may have come in', function () { this.UpdateManager.promises.continueProcessingUpdatesWithLock - .calledWith(this.project_id, this.doc_id) + .calledWith(this.project_id) .should.equal(true) }) }) - describe('when processOutstandingProjectUpdates returns an error', function () { + describe('when fetchAndApplyProjectUpdates returns an error', function () { beforeEach(async function () { this.error = new Error('Something went wrong') - this.UpdateManager.promises.processOutstandingProjectUpdates = sinon + this.UpdateManager.promises.fetchAndApplyProjectUpdates = sinon .stub() .rejects(this.error) await expect( this.UpdateManager.promises.processOutstandingUpdatesWithLock( - this.project_id, - this.doc_id + this.project_id ) ).to.be.rejectedWith(this.error) }) @@ -224,161 +193,6 @@ describe('UpdateManager', 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.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 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 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() @@ -460,36 +274,30 @@ describe('UpdateManager', function () { describe('continueProcessingUpdatesWithLock', function () { describe('when there are outstanding updates', function () { beforeEach(async function () { - this.UpdateManager.promises.getProjectPendingUpdatesLength = sinon - .stub() - .resolves(3) + this.RealTimeRedisManager.promises.getProjectUpdatesLength.resolves(3) this.UpdateManager.promises.processOutstandingUpdatesWithLock = sinon .stub() .resolves() await this.UpdateManager.promises.continueProcessingUpdatesWithLock( - this.project_id, - this.doc_id + this.project_id ) }) it('should process the outstanding updates', function () { this.UpdateManager.promises.processOutstandingUpdatesWithLock - .calledWith(this.project_id, this.doc_id) + .calledWith(this.project_id) .should.equal(true) }) }) describe('when there are no outstanding updates', function () { beforeEach(async function () { - this.UpdateManager.promises.getProjectPendingUpdatesLength = sinon - .stub() - .resolves(0) + this.RealTimeRedisManager.promises.getProjectUpdatesLength.resolves(0) this.UpdateManager.promises.processOutstandingUpdatesWithLock = sinon .stub() .resolves() await this.UpdateManager.promises.continueProcessingUpdatesWithLock( - this.project_id, - this.doc_id + this.project_id ) }) @@ -501,111 +309,6 @@ 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 () { - this.updates = [{ p: 1, t: 'foo' }] - this.updatedDocLines = ['updated', 'lines'] - this.version = 34 - this.RealTimeRedisManager.promises.getPendingUpdatesForDoc.resolves( - this.updates - ) - this.UpdateManager.promises.applyUpdate = sinon.stub().resolves() - await this.UpdateManager.promises.fetchAndApplyUpdates( - this.project_id, - this.doc_id - ) - }) - - it('should get the pending updates', function () { - this.RealTimeRedisManager.promises.getPendingUpdatesForDoc - .calledWith(this.doc_id) - .should.equal(true) - }) - - it('should apply the updates', function () { - this.updates.map(update => - this.UpdateManager.promises.applyUpdate - .calledWith(this.project_id, this.doc_id, update) - .should.equal(true) - ) - }) - }) - - describe('when there are no updates', function () { - beforeEach(async function () { - this.updates = [] - this.RealTimeRedisManager.promises.getPendingUpdatesForDoc.resolves( - this.updates - ) - this.UpdateManager.promises.applyUpdate = sinon.stub().resolves() - await this.UpdateManager.promises.fetchAndApplyUpdates( - this.project_id, - this.doc_id - ) - }) - - it('should not call applyUpdate', function () { - this.UpdateManager.promises.applyUpdate.called.should.equal(false) - }) - }) - }) - describe('applyUpdate', function () { beforeEach(function () { this.updateMeta = { user_id: 'last-author-fake-id' } @@ -1158,7 +861,7 @@ describe('UpdateManager', function () { this.UpdateManager.promises.continueProcessingUpdatesWithLock = sinon .stub() .resolves() - this.UpdateManager.promises.processOutstandingProjectUpdates = sinon + this.UpdateManager.promises.fetchAndApplyProjectUpdates = sinon .stub() .resolves() this.response = await this.UpdateManager.promises.lockUpdatesAndDo( @@ -1182,10 +885,8 @@ describe('UpdateManager', function () { }) 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.projectLockValue + this.UpdateManager.promises.fetchAndApplyProjectUpdates.should.have.been.calledWith( + this.project_id ) }) @@ -1210,15 +911,15 @@ describe('UpdateManager', function () { it('should continue processing updates', function () { this.UpdateManager.promises.continueProcessingUpdatesWithLock - .calledWith(this.project_id, this.doc_id) + .calledWith(this.project_id) .should.equal(true) }) }) - describe('when processOutstandingProjectUpdates returns an error', function () { + describe('when fetchAndApplyProjectUpdates returns an error', function () { beforeEach(async function () { this.error = new Error('Something went wrong') - this.UpdateManager.promises.processOutstandingProjectUpdates = sinon + this.UpdateManager.promises.fetchAndApplyProjectUpdates = sinon .stub() .rejects(this.error) await expect( @@ -1241,7 +942,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.processOutstandingProjectUpdates = sinon + this.UpdateManager.promises.fetchAndApplyProjectUpdates = sinon .stub() .resolves() this.method = sinon.stub().rejects(this.error) diff --git a/services/real-time/app/js/DocumentUpdaterManager.js b/services/real-time/app/js/DocumentUpdaterManager.js index 2c3d10d9fe..5ae05525b5 100644 --- a/services/real-time/app/js/DocumentUpdaterManager.js +++ b/services/real-time/app/js/DocumentUpdaterManager.js @@ -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 }), diff --git a/services/real-time/app/js/WebApiManager.js b/services/real-time/app/js/WebApiManager.js index 48f150e6aa..61e1670ce4 100644 --- a/services/real-time/app/js/WebApiManager.js +++ b/services/real-time/app/js/WebApiManager.js @@ -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, diff --git a/services/real-time/app/js/WebsocketController.js b/services/real-time/app/js/WebsocketController.js index 07871d7413..8e42031e69 100644 --- a/services/real-time/app/js/WebsocketController.js +++ b/services/real-time/app/js/WebsocketController.js @@ -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') diff --git a/services/real-time/config/settings.defaults.cjs b/services/real-time/config/settings.defaults.cjs index 089f2f8695..58db305429 100644 --- a/services/real-time/config/settings.defaults.cjs +++ b/services/real-time/config/settings.defaults.cjs @@ -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}` + - // `:` marker) - // phase 2: write the per-project queue (`PendingProjectUpdates:{project_id}` - // + bare `` 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) diff --git a/services/real-time/test/acceptance/js/ApplyUpdateTests.js b/services/real-time/test/acceptance/js/ApplyUpdateTests.js index 2901a78f2a..b0a8b3cda0 100644 --- a/services/real-time/test/acceptance/js/ApplyUpdateTests.js +++ b/services/real-time/test/acceptance/js/ApplyUpdateTests.js @@ -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) diff --git a/services/real-time/test/acceptance/js/MatrixTests.js b/services/real-time/test/acceptance/js/MatrixTests.js index 3adf579f92..69ae06950e 100644 --- a/services/real-time/test/acceptance/js/MatrixTests.js +++ b/services/real-time/test/acceptance/js/MatrixTests.js @@ -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) }) diff --git a/services/real-time/test/acceptance/js/helpers/MockWebServer.js b/services/real-time/test/acceptance/js/helpers/MockWebServer.js index a46d262115..bbcbb630c1 100644 --- a/services/real-time/test/acceptance/js/helpers/MockWebServer.js +++ b/services/real-time/test/acceptance/js/helpers/MockWebServer.js @@ -74,8 +74,6 @@ export default MockWebServer = { isRestrictedUser: !!userMetadata?.isRestrictedUser, isTokenMember: !!userMetadata?.isTokenMember, isInvitedMember: !!userMetadata?.isInvitedMember, - usePerProjectPendingUpdates: - !!userMetadata?.usePerProjectPendingUpdates, }) } ) diff --git a/services/real-time/test/unit/js/DocumentUpdaterController.test.js b/services/real-time/test/unit/js/DocumentUpdaterController.test.js index bd2ad4cfb8..beb7c03563 100644 --- a/services/real-time/test/unit/js/DocumentUpdaterController.test.js +++ b/services/real-time/test/unit/js/DocumentUpdaterController.test.js @@ -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, }, diff --git a/services/real-time/test/unit/js/DocumentUpdaterManager.test.js b/services/real-time/test/unit/js/DocumentUpdaterManager.test.js index e4ae47ebc4..5ad928e422 100644 --- a/services/real-time/test/unit/js/DocumentUpdaterManager.test.js +++ b/services/real-time/test/unit/js/DocumentUpdaterManager.test.js @@ -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) diff --git a/services/real-time/test/unit/js/WebApiManager.test.js b/services/real-time/test/unit/js/WebApiManager.test.js index bb8f219ef6..3b0667854c 100644 --- a/services/real-time/test/unit/js/WebApiManager.test.js +++ b/services/real-time/test/unit/js/WebApiManager.test.js @@ -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, } ) }) diff --git a/services/real-time/test/unit/js/WebsocketController.test.js b/services/real-time/test/unit/js/WebsocketController.test.js index b4c449caae..8c6cc7ae48 100644 --- a/services/real-time/test/unit/js/WebsocketController.test.js +++ b/services/real-time/test/unit/js/WebsocketController.test.js @@ -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, diff --git a/services/web/app/src/Features/Editor/EditorHttpController.mjs b/services/web/app/src/Features/Editor/EditorHttpController.mjs index d4387c64ec..d98968f417 100644 --- a/services/web/app/src/Features/Editor/EditorHttpController.mjs +++ b/services/web/app/src/Features/Editor/EditorHttpController.mjs @@ -7,7 +7,6 @@ import Metrics from '@overleaf/metrics' import CollaboratorsInviteGetter from '../Collaborators/CollaboratorsInviteGetter.mjs' import PrivilegeLevels from '../Authorization/PrivilegeLevels.mjs' import SessionManager from '../Authentication/SessionManager.mjs' -import SplitTestHandler from '../SplitTests/SplitTestHandler.mjs' import Errors from '../Errors/Errors.js' import { expressify } from '@overleaf/promise-utils' import Settings from '@overleaf/settings' @@ -68,22 +67,12 @@ async function joinProject(req, res, next) { ) } - let usePerProjectPendingUpdates = false - if (userId) { - usePerProjectPendingUpdates = - await SplitTestHandler.promises.featureFlagEnabledForUser( - userId, - 'per-project-pending-updates' - ) - } - res.json({ project, privilegeLevel, isRestrictedUser, isTokenMember, isInvitedMember, - usePerProjectPendingUpdates, }) } diff --git a/services/web/scripts/fix_oversized_docs.mjs b/services/web/scripts/fix_oversized_docs.mjs index d3c75e1de8..b218ac3e20 100644 --- a/services/web/scripts/fix_oversized_docs.mjs +++ b/services/web/scripts/fix_oversized_docs.mjs @@ -15,7 +15,7 @@ const redis = RedisWrapper.createClient(Settings.redis.web) function parseArgs() { const args = minimist(process.argv.slice(2), { - boolean: ['commit', 'ignore-ranges'], + boolean: ['commit', 'ignore-ranges', 'ignore-pending-updates'], }) const projectIds = args._ @@ -23,20 +23,22 @@ function parseArgs() { console.log(`Usage: ${process.argv[1]} [OPTS] PROJECT_ID Options: - --commit Actually convert oversized docs to binary files - --max-doc-size Size over which docs are converted to binary files - --ignore-ranges Convert docs even if they contain ranges + --commit Actually convert oversized docs to binary files + --max-doc-size Size over which docs are converted to binary files + --ignore-ranges Convert docs even if they contain ranges + --ignore-pending-updates Process projects even if they have pending updates `) process.exit(0) } const commit = args.commit const ignoreRanges = args['ignore-ranges'] + const ignorePendingUpdates = args['ignore-pending-updates'] const maxDocSize = args['max-doc-size'] ? parseInt(args['max-doc-size'], 10) : 2 * 1024 * 1024 - return { projectIds, commit, ignoreRanges, maxDocSize } + return { projectIds, commit, ignoreRanges, ignorePendingUpdates, maxDocSize } } async function main() { @@ -49,12 +51,36 @@ async function main() { } async function processProject(projectId) { + await checkPendingUpdates(projectId) const docIds = await getDocIds(projectId) for (const docId of docIds) { await processDoc(projectId, docId) } } +// A doc's pending updates live in the shared +// `PendingProjectUpdates:{projectId}` list (on the document-updater redis). +// They would be lost when converting the doc to a binary file, so refuse to +// process a project that has pending updates - flush/drain the project first, +// or override with --ignore-pending-updates. +async function checkPendingUpdates(projectId) { + const pendingUpdates = await redis.llen( + `PendingProjectUpdates:{${projectId}}` + ) + if (pendingUpdates === 0) { + return + } + if (opts.ignorePendingUpdates) { + console.log( + `Ignoring ${pendingUpdates} pending update(s) in project ${projectId}` + ) + } else { + throw new Error( + `project ${projectId} has ${pendingUpdates} pending update(s) in redis - flush/drain the project first, or re-run with --ignore-pending-updates` + ) + } +} + async function processDoc(projectId, docId) { const doc = await getDoc(projectId, docId) const size = doc.lines.reduce((sum, line) => sum + line.length + 1, 0) @@ -148,15 +174,12 @@ async function deleteDocFromRedis(projectId, docId) { `UnflushedTime:{${docId}}`, `Pathname:{${docId}}`, `ProjectHistoryId:{${docId}}`, - `PendingUpdates:{${docId}}`, `lastUpdatedAt:{${docId}}`, `lastUpdatedBy:{${docId}}` ) - // NOTE: once the update-queue migration reaches phase >= 2, a doc's pending - // updates live in the shared `PendingProjectUpdates:{projectId}` list (on the - // document-updater redis) rather than `PendingUpdates:{docId}`. That list is - // per-project, so it cannot be cleared here without dropping other docs' - // updates - flush/drain the project before running this script. + // NOTE: the shared `PendingProjectUpdates:{projectId}` list is per-project + // and cannot be cleared here without dropping other docs' updates - see + // checkPendingUpdates above. await redis.srem(`DocsIn:{${projectId}}`, docId) } diff --git a/services/web/scripts/recover_docs_from_redis.mjs b/services/web/scripts/recover_docs_from_redis.mjs index 3c223be3c0..286266478d 100644 --- a/services/web/scripts/recover_docs_from_redis.mjs +++ b/services/web/scripts/recover_docs_from_redis.mjs @@ -14,13 +14,15 @@ const redis = RedisWrapper.createClient(Settings.redis.web) function parseArgs() { const args = minimist(process.argv.slice(2), { - boolean: ['commit'], + boolean: ['commit', 'ignore-pending-updates'], }) const commit = args.commit const deleteOversized = args['delete-oversized'] + const ignorePendingUpdates = args['ignore-pending-updates'] return { commit, deleteOversized, + ignorePendingUpdates, maxDocSize: 2 * 1024 * 1024, } } @@ -79,6 +81,28 @@ async function processDoc(docId) { ) return } + // A doc's pending updates live in the shared + // `PendingProjectUpdates:{projectId}` list (on the document-updater redis). + // Restoring the doc - or deleting it when oversized - would drop them, so + // skip the doc while the project has pending updates: flush/drain the + // project first, or override with --ignore-pending-updates. + const pendingUpdates = await redis.llen( + `PendingProjectUpdates:{${projectId}}` + ) + if (pendingUpdates > 0) { + if (opts.ignorePendingUpdates) { + logger.warn( + { docId, projectId, pendingUpdates }, + 'ignoring pending updates for project' + ) + } else { + logger.warn( + { docId, projectId, pendingUpdates }, + 'project has pending updates - skipping doc (flush/drain the project or use --ignore-pending-updates)' + ) + return + } + } // if the doc is too big we will need to convert it to a file, skip it for now // or delete the doc if the --delete-oversized option is used const size = doc.lines.reduce((sum, line) => sum + line.length + 1, 0) @@ -166,15 +190,12 @@ async function deleteDocFromRedis(projectId, docId) { `UnflushedTime:{${docId}}`, `Pathname:{${docId}}`, `ProjectHistoryId:{${docId}}`, - `PendingUpdates:{${docId}}`, `lastUpdatedAt:{${docId}}`, `lastUpdatedBy:{${docId}}` ) - // NOTE: once the update-queue migration reaches phase >= 2, a doc's pending - // updates live in the shared `PendingProjectUpdates:{projectId}` list (on the - // document-updater redis) rather than `PendingUpdates:{docId}`. That list is - // per-project, so it cannot be cleared here without dropping other docs' - // updates - flush/drain the project before running this script. + // NOTE: the shared `PendingProjectUpdates:{projectId}` list is per-project + // and cannot be cleared here without dropping other docs' updates - the + // pending-updates check in processDoc guards against this. await redis.srem(`DocsIn:{${projectId}}`, docId) } diff --git a/services/web/test/unit/src/Editor/EditorHttpController.test.mjs b/services/web/test/unit/src/Editor/EditorHttpController.test.mjs index e792b57a18..5b8ea41c8e 100644 --- a/services/web/test/unit/src/Editor/EditorHttpController.test.mjs +++ b/services/web/test/unit/src/Editor/EditorHttpController.test.mjs @@ -305,7 +305,6 @@ describe('EditorHttpController', function () { isRestrictedUser: false, isTokenMember: false, isInvitedMember: true, - usePerProjectPendingUpdates: false, }) }) @@ -363,7 +362,6 @@ describe('EditorHttpController', function () { isRestrictedUser: true, isTokenMember: false, isInvitedMember: false, - usePerProjectPendingUpdates: false, }) }) }) @@ -420,14 +418,8 @@ describe('EditorHttpController', function () { isRestrictedUser: true, isTokenMember: false, isInvitedMember: false, - usePerProjectPendingUpdates: false, }) }) - - it('should not fetch a split test assignment', function (ctx) { - expect(ctx.SplitTestHandler.promises.featureFlagEnabledForUser).not.to - .have.been.called - }) }) describe('with a token access user', function () { @@ -460,36 +452,10 @@ describe('EditorHttpController', function () { isRestrictedUser: false, isTokenMember: true, isInvitedMember: false, - usePerProjectPendingUpdates: false, }) }) }) - describe('with the per-project-pending-updates flag enabled', function () { - beforeEach(async function (ctx) { - ctx.SplitTestHandler.promises.featureFlagEnabledForUser.resolves(true) - await new Promise(resolve => { - ctx.res.callback = resolve - ctx.EditorHttpController.joinProject(ctx.req, ctx.res) - }) - }) - - it('should fetch the assignment for the joining user', function (ctx) { - expect( - ctx.SplitTestHandler.promises.featureFlagEnabledForUser - ).to.have.been.calledWith( - ctx.user._id.toString(), - 'per-project-pending-updates' - ) - }) - - it('should return usePerProjectPendingUpdates=true', function (ctx) { - expect(ctx.res.json).toHaveBeenCalledWith( - expect.objectContaining({ usePerProjectPendingUpdates: true }) - ) - }) - }) - describe('when project is not found', function () { beforeEach(async function (ctx) { ctx.ProjectGetter.promises.getProject.resolves(null)