From 324f011d8055f836f186d2c8596336e761767516 Mon Sep 17 00:00:00 2001 From: Jakob Ackermann Date: Mon, 20 Jul 2026 12:28:25 +0200 Subject: [PATCH] [document-updater] add script for flushing/discarding stuck docs (#35603) GitOrigin-RevId: 099e1e0a38515b35decbdc608acd75f59de0ec6f --- .../document-updater/app/js/ProjectFlusher.js | 1 + .../flush_docs_with_pending_updates.js | 281 +++++++++++++ .../js/FlushDocsWithPendingUpdatesTests.js | 384 ++++++++++++++++++ 3 files changed, 666 insertions(+) create mode 100644 services/document-updater/scripts/flush_docs_with_pending_updates.js create mode 100644 services/document-updater/test/acceptance/js/FlushDocsWithPendingUpdatesTests.js diff --git a/services/document-updater/app/js/ProjectFlusher.js b/services/document-updater/app/js/ProjectFlusher.js index d6b8c5b133..fb6687417a 100644 --- a/services/document-updater/app/js/ProjectFlusher.js +++ b/services/document-updater/app/js/ProjectFlusher.js @@ -96,6 +96,7 @@ async function flushAllProjects(options) { } module.exports = { + _getKeys, _extractIds, promises: { flushAllProjects, diff --git a/services/document-updater/scripts/flush_docs_with_pending_updates.js b/services/document-updater/scripts/flush_docs_with_pending_updates.js new file mode 100644 index 0000000000..1be0f3c6ac --- /dev/null +++ b/services/document-updater/scripts/flush_docs_with_pending_updates.js @@ -0,0 +1,281 @@ +// Flush docs that were left with a stuck PendingUpdates queue in redis after +// their project was flushed (and possibly hard deleted since). +// +// For each doc with a PendingUpdates queue, the project id is looked up in +// db.docs, then the pending updates are processed and the doc is flushed and +// removed from redis. Docs whose project cannot be found (e.g. hard deleted +// projects) are logged and can be discarded with --discard-hard-deleted. + +const minimist = require('minimist') +const Settings = require('@overleaf/settings') +const OError = require('@overleaf/o-error') +const DocumentManager = require('../app/js/DocumentManager') +const HistoryManager = require('../app/js/HistoryManager') +const ProjectFlusher = require('../app/js/ProjectFlusher') +const { rclient } = require('../app/js/RedisManager') +const { db, ObjectId, mongoClient } = require('../app/js/mongodb') + +const keys = Settings.redis.documentupdater.key_schema + +const summary = { + totalDocs: 0, + flushedDocs: 0, + missingDocs: 0, + discardedDocs: 0, + erroredDocs: 0, +} + +function printSummary() { + console.log( + `Processed ${summary.totalDocs} docs with pending updates: flushed=${summary.flushedDocs} missing=${summary.missingDocs} discarded=${summary.discardedDocs} errored=${summary.erroredDocs}` + ) +} + +function usage() { + console.error(` +Usage: node scripts/flush_docs_with_pending_updates.js [options] + +Options: + --estimated-docs Required. Estimated number of documents in db.docs, + used as a safety check that the right database is + consulted for the doc lookups. The script aborts + unless the estimated collection count is within 20% + of this value. + --log-unflushed When the db.docs lookup failed for a doc, log its + pending updates and, if the doc has unflushed + changes, all its doc keys + (default: true, disable via --log-unflushed=false) + --discard-hard-deleted Discard the stuck queue and the other doc keys of + docs whose db.docs lookup failed (default: false) + --docId Process a single doc instead of scanning redis + --projectId Override the project id for --docId, skipping the + db.docs lookup + --dry-run Only scan and log, without flushing, deleting or + discarding anything (default: false) + --limit Maximum number of PendingUpdates keys to scan + (default: 100000) + --help Show this help message +`) +} + +async function getDocIds(options) { + if (options.docId) { + return [options.docId] + } + const docKeys = await ProjectFlusher._getKeys( + keys.pendingUpdates({ doc_id: '*' }), + options.limit + ) + return ProjectFlusher._extractIds(docKeys) +} + +async function processDoc(docId, options) { + let projectId = options.projectId + if (!projectId) { + const doc = await db.docs.findOne( + { _id: new ObjectId(docId) }, + { projection: { project_id: 1 } } + ) + projectId = doc?.project_id?.toString() + } + + if (projectId) { + await flushDoc(projectId, docId, options) + } else { + await handleMissingDoc(docId, options) + } +} + +async function flushDoc(projectId, docId, options) { + const queueLength = await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + if (options.dryRun) { + console.log( + `Would flush doc ${docId} in project ${projectId} with ${queueLength} pending updates` + ) + return + } + console.log( + `Flushing doc ${docId} in project ${projectId} with ${queueLength} pending updates` + ) + await DocumentManager.promises.flushAndDeleteDocWithLock(projectId, docId, {}) + await HistoryManager.promises.flushProjectChanges(projectId, {}) + const remaining = await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + if (remaining > 0) { + console.error( + `Doc ${docId} in project ${projectId} still has ${remaining} pending updates after flushing` + ) + summary.erroredDocs++ + } else { + summary.flushedDocs++ + } +} + +async function handleMissingDoc(docId, options) { + summary.missingDocs++ + let message = `Doc ${docId} not found in db.docs` + + if (options.logUnflushed) { + // Log the pending updates: they are the reason this doc is processed and + // would be lost when discarding it. These can exist independently of any + // unflushed doc content. + const pendingUpdates = await rclient.lrange( + keys.pendingUpdates({ doc_id: docId }), + 0, + -1 + ) + message += `, pending updates: ${JSON.stringify(pendingUpdates)}` + + // In addition, log the unflushed doc content if there is any. + const unflushedTime = await rclient.get( + keys.unflushedTime({ doc_id: docId }) + ) + if (unflushedTime) { + const doc = { + unflushedTime, + projectId: await rclient.get(keys.projectKey({ doc_id: docId })), + lines: await rclient.get(keys.docLines({ doc_id: docId })), + version: await rclient.get(keys.docVersion({ doc_id: docId })), + hash: await rclient.get(keys.docHash({ doc_id: docId })), + ranges: await rclient.get(keys.ranges({ doc_id: docId })), + pathname: await rclient.get(keys.pathname({ doc_id: docId })), + projectHistoryId: await rclient.get( + keys.projectHistoryId({ doc_id: docId }) + ), + lastUpdatedAt: await rclient.get(keys.lastUpdatedAt({ doc_id: docId })), + lastUpdatedBy: await rclient.get(keys.lastUpdatedBy({ doc_id: docId })), + resolvedCommentIds: await rclient.smembers( + keys.resolvedCommentIds({ doc_id: docId }) + ), + } + message += `, unflushed doc keys: ${JSON.stringify(doc)}` + } + } + console.warn(message) + + if (options.discardHardDeleted && !options.dryRun) { + await discardDoc(docId) + console.warn(`Discarded redis keys of doc ${docId}`) + summary.discardedDocs++ + } +} + +async function discardDoc(docId) { + // Purge the project-level entries first: in case the script crashes mid-way, + // a re-run can still resolve the project id from the ProjectId doc key. + const projectId = await rclient.get(keys.projectKey({ doc_id: docId })) + if (projectId) { + await rclient.srem(keys.docsInProject({ project_id: projectId }), docId) + } + await rclient.srem(keys.historyRangesSupport(), docId) + await rclient.del( + keys.pendingUpdates({ doc_id: docId }), + keys.docLines({ doc_id: docId }), + keys.projectKey({ doc_id: docId }), + keys.docVersion({ doc_id: docId }), + keys.docHash({ doc_id: docId }), + keys.ranges({ doc_id: docId }), + keys.pathname({ doc_id: docId }), + keys.projectHistoryId({ doc_id: docId }), + keys.unflushedTime({ doc_id: docId }), + keys.lastUpdatedBy({ doc_id: docId }), + keys.lastUpdatedAt({ doc_id: docId }), + keys.resolvedCommentIds({ doc_id: docId }), + keys.docOps({ doc_id: docId }) + ) +} + +async function main() { + const argv = minimist(process.argv.slice(2), { + boolean: ['log-unflushed', 'discard-hard-deleted', 'dry-run', 'help'], + string: ['docId', 'projectId'], + default: { limit: 100_000, 'log-unflushed': true }, + }) + + if (argv.help) { + usage() + return 0 + } + + const options = { + estimatedDocs: argv['estimated-docs'], + logUnflushed: argv['log-unflushed'], + discardHardDeleted: argv['discard-hard-deleted'], + docId: argv.docId, + projectId: argv.projectId, + dryRun: argv['dry-run'], + limit: argv.limit, + } + + if (!Number.isInteger(options.estimatedDocs) || options.estimatedDocs <= 0) { + console.error('--estimated-docs must be set to a positive integer') + usage() + return 1 + } + if (options.docId && !ObjectId.isValid(options.docId)) { + console.error('--docId is not a valid object id') + usage() + return 1 + } + if (options.projectId && !options.docId) { + console.error('--projectId can only be used together with --docId') + usage() + return 1 + } + if (options.projectId && !ObjectId.isValid(options.projectId)) { + console.error('--projectId is not a valid object id') + usage() + return 1 + } + + const actualDocs = await db.docs.estimatedDocumentCount() + if ( + actualDocs < 0.8 * options.estimatedDocs || + actualDocs > 1.2 * options.estimatedDocs + ) { + console.error( + `Aborting: the estimated db.docs count of ${actualDocs} is not within 20% of --estimated-docs=${options.estimatedDocs}, is this the right database?` + ) + return 1 + } + + const docIds = await getDocIds(options) + const limitReached = !options.docId && docIds.length >= options.limit + for (const docId of docIds) { + summary.totalDocs++ + try { + await processDoc(docId, options) + } catch (error) { + console.error(`Error processing doc ${docId}`) + console.error(OError.getFullStack(error)) + summary.erroredDocs++ + } + } + + if (limitReached) { + console.error( + `Hit the scan limit of ${options.limit} keys; there may be more docs with pending updates. Re-run to process the rest.` + ) + return 2 + } + if (options.dryRun) { + return summary.totalDocs > 0 ? 2 : 0 + } + if (summary.erroredDocs > 0 || summary.missingDocs > summary.discardedDocs) { + return 2 + } + return 0 +} + +main() + .then(code => { + printSummary() + rclient.quit() + mongoClient.close() + process.exit(code) + }) + .catch(error => { + printSummary() + console.error(OError.getFullStack(error)) + console.error(OError.getFullInfo(error)) + process.exit(1) + }) diff --git a/services/document-updater/test/acceptance/js/FlushDocsWithPendingUpdatesTests.js b/services/document-updater/test/acceptance/js/FlushDocsWithPendingUpdatesTests.js new file mode 100644 index 0000000000..eee6c0f932 --- /dev/null +++ b/services/document-updater/test/acceptance/js/FlushDocsWithPendingUpdatesTests.js @@ -0,0 +1,384 @@ +const MockWebApi = require('./helpers/MockWebApi') +const DocUpdaterClient = require('./helpers/DocUpdaterClient') +const DocUpdaterApp = require('./helpers/DocUpdaterApp') +const { promisify } = require('node:util') +const { exec } = require('node:child_process') +const { expect } = require('chai') +const Settings = require('@overleaf/settings') +const { db, ObjectId } = require('../../../app/js/mongodb') + +const rclient = require('@overleaf/redis-wrapper').createClient( + Settings.redis.documentupdater +) +const keys = Settings.redis.documentupdater.key_schema + +describe('FlushDocsWithPendingUpdates', function () { + beforeEach(async function () { + await DocUpdaterApp.ensureRunning() + await rclient.flushall() + await db.docs.deleteMany({}) + // Make sure db.docs is never empty, so that a valid --estimated-docs + // value exists even in the scenarios that do not insert a doc. + await db.docs.insertOne({ + _id: new ObjectId(), + project_id: new ObjectId(), + }) + }) + + async function runScript(args) { + let result + try { + result = await promisify(exec)( + ['node', 'scripts/flush_docs_with_pending_updates.js'] + .concat(args) + .join(' ') + ) + } catch (error) { + // includes details like exit code, stdErr and stdOut + return error + } + result.code = 0 + return result + } + + async function getEstimatedDocs() { + return await db.docs.estimatedDocumentCount() + } + + async function createStuckDoc({ insertIntoMongo }) { + const projectId = DocUpdaterClient.randomId() + const docId = DocUpdaterClient.randomId() + MockWebApi.insertDoc(projectId, docId, { + lines: ['one', 'two', 'three'], + version: 1, + }) + if (insertIntoMongo) { + await db.docs.insertOne({ + _id: new ObjectId(docId), + project_id: new ObjectId(projectId), + }) + } + // Simulate the race condition: the doc is no longer loaded in redis, but + // an update is still queued. + const update = { + doc: docId, + op: [{ i: 'foo ', p: 0 }], + v: 1, + meta: { user_id: DocUpdaterClient.randomId(), ts: Date.now() }, + } + await rclient.rpush( + keys.pendingUpdates({ doc_id: docId }), + JSON.stringify(update) + ) + return { projectId, docId } + } + + describe('with a stuck doc that exists in mongo', function () { + let projectId, docId + beforeEach(async function () { + ;({ projectId, docId } = await createStuckDoc({ insertIntoMongo: true })) + }) + + it('should apply the pending updates and flush and delete the doc', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + ]) + expect(result.code).to.equal(0) + expect(result.stdout).to.include( + `Flushing doc ${docId} in project ${projectId} with 1 pending updates` + ) + expect(result.stdout).to.include( + 'Processed 1 docs with pending updates: flushed=1 missing=0 discarded=0 errored=0' + ) + + expect( + await rclient.exists(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(0) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + + const doc = await MockWebApi.getDocument(projectId, docId) + expect(doc.lines).to.deep.equal(['foo one', 'two', 'three']) + expect(doc.version).to.equal(2) + }) + + it('should not change anything in dry-run mode', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + '--dry-run', + ]) + expect(result.code).to.equal(2) + expect(result.stdout).to.include( + `Would flush doc ${docId} in project ${projectId} with 1 pending updates` + ) + + expect( + await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(1) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + const doc = await MockWebApi.getDocument(projectId, docId) + expect(doc.lines).to.deep.equal(['one', 'two', 'three']) + expect(doc.version).to.equal(1) + }) + + describe('with a second stuck doc', function () { + let otherDocId + beforeEach(async function () { + ;({ docId: otherDocId } = await createStuckDoc({ + insertIntoMongo: true, + })) + }) + + it('should only process the doc passed via --docId', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + `--docId=${docId}`, + ]) + expect(result.code).to.equal(0) + expect(result.stdout).to.include( + 'Processed 1 docs with pending updates: flushed=1 missing=0 discarded=0 errored=0' + ) + + expect( + await rclient.exists(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(0) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal( + 0 + ) + expect( + await rclient.llen(keys.pendingUpdates({ doc_id: otherDocId })) + ).to.equal(1) + expect( + await rclient.exists(keys.docLines({ doc_id: otherDocId })) + ).to.equal(0) + }) + + it('should exit non-zero when hitting the scan limit', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + '--limit=1', + ]) + expect(result.code).to.equal(2) + expect(result.stderr).to.include('Hit the scan limit of 1 keys') + }) + }) + }) + + describe('with a stuck doc whose project was hard deleted', function () { + let projectId, docId + beforeEach(async function () { + ;({ projectId, docId } = await createStuckDoc({ + insertIntoMongo: false, + })) + }) + + it('should log the doc id and leave the queue in place', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + ]) + expect(result.code).to.equal(2) + expect(result.stderr).to.include( + `Doc ${docId} not found in db.docs, pending updates:` + ) + expect(result.stderr).to.include('foo ') + expect(result.stderr).to.not.include('unflushed doc keys:') + expect(result.stdout).to.include( + 'Processed 1 docs with pending updates: flushed=0 missing=1 discarded=0 errored=0' + ) + + expect( + await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(1) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + }) + + describe('with unflushed changes', function () { + beforeEach(async function () { + await rclient.set(keys.unflushedTime({ doc_id: docId }), Date.now()) + await rclient.set(keys.projectKey({ doc_id: docId }), projectId) + await rclient.set(keys.docLines({ doc_id: docId }), '["unflushed"]') + await rclient.set(keys.docVersion({ doc_id: docId }), 2) + }) + + it('should log all the doc keys by default', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + ]) + expect(result.code).to.equal(2) + expect(result.stderr).to.include( + `Doc ${docId} not found in db.docs, pending updates:` + ) + expect(result.stderr).to.include('unflushed doc keys:') + expect(result.stderr).to.include('unflushed') + expect(result.stderr).to.include('foo ') + }) + + it('should not log the doc content with --log-unflushed=false', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + '--log-unflushed=false', + ]) + expect(result.code).to.equal(2) + expect(result.stderr).to.include(`Doc ${docId} not found in db.docs`) + expect(result.stderr).to.not.include('pending updates:') + expect(result.stderr).to.not.include('unflushed doc keys:') + }) + + it('should log but not discard the doc keys in dry-run mode', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + '--discard-hard-deleted', + '--dry-run', + ]) + expect(result.code).to.equal(2) + expect(result.stderr).to.include('unflushed doc keys:') + expect(result.stderr).to.not.include( + `Discarded redis keys of doc ${docId}` + ) + + expect( + await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(1) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal( + 1 + ) + }) + }) + + it('should not log the doc keys when the lines are flushed', async function () { + // Doc content is present in redis but already flushed: no unflushed time. + await rclient.set(keys.projectKey({ doc_id: docId }), projectId) + await rclient.set(keys.docLines({ doc_id: docId }), '["flushed"]') + await rclient.set(keys.docVersion({ doc_id: docId }), 2) + + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + ]) + expect(result.code).to.equal(2) + expect(result.stderr).to.include( + `Doc ${docId} not found in db.docs, pending updates:` + ) + expect(result.stderr).to.not.include('unflushed doc keys:') + }) + + it('should remove all the doc keys with --discard-hard-deleted', async function () { + // Simulate leftovers of a partially loaded doc. + await rclient.set(keys.projectKey({ doc_id: docId }), projectId) + await rclient.set(keys.docLines({ doc_id: docId }), '["one"]') + await rclient.sadd(keys.docsInProject({ project_id: projectId }), docId) + await rclient.sadd(keys.historyRangesSupport(), docId) + + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + '--discard-hard-deleted', + ]) + expect(result.code).to.equal(0) + expect(result.stderr).to.include(`Discarded redis keys of doc ${docId}`) + expect(result.stdout).to.include( + 'Processed 1 docs with pending updates: flushed=0 missing=1 discarded=1 errored=0' + ) + + expect( + await rclient.exists(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(0) + expect(await rclient.exists(keys.projectKey({ doc_id: docId }))).to.equal( + 0 + ) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + expect( + await rclient.sismember( + keys.docsInProject({ project_id: projectId }), + docId + ) + ).to.equal(0) + expect( + await rclient.sismember(keys.historyRangesSupport(), docId) + ).to.equal(0) + }) + + it('should not discard anything in dry-run mode', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + '--discard-hard-deleted', + '--dry-run', + ]) + expect(result.code).to.equal(2) + expect( + await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(1) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + }) + + it('should flush the doc with a --projectId override', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + `--docId=${docId}`, + `--projectId=${projectId}`, + ]) + expect(result.code).to.equal(0) + expect(result.stdout).to.include( + 'Processed 1 docs with pending updates: flushed=1 missing=0 discarded=0 errored=0' + ) + + expect( + await rclient.exists(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(0) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + const doc = await MockWebApi.getDocument(projectId, docId) + expect(doc.lines).to.deep.equal(['foo one', 'two', 'three']) + expect(doc.version).to.equal(2) + }) + }) + + describe('safety measures', function () { + let docId + beforeEach(async function () { + ;({ docId } = await createStuckDoc({ insertIntoMongo: true })) + }) + + it('should abort without --estimated-docs', async function () { + const result = await runScript([]) + expect(result.code).to.equal(1) + expect(result.stderr).to.include( + '--estimated-docs must be set to a positive integer' + ) + expect( + await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(1) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + }) + + it('should abort when --estimated-docs is far off', async function () { + const estimatedDocs = (await getEstimatedDocs()) * 10 + const result = await runScript([`--estimated-docs=${estimatedDocs}`]) + expect(result.code).to.equal(1) + expect(result.stderr).to.include('is this the right database?') + expect( + await rclient.llen(keys.pendingUpdates({ doc_id: docId })) + ).to.equal(1) + expect(await rclient.exists(keys.docLines({ doc_id: docId }))).to.equal(0) + }) + + it('should abort when --projectId is used without --docId', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + `--projectId=${new ObjectId().toString()}`, + ]) + expect(result.code).to.equal(1) + expect(result.stderr).to.include( + '--projectId can only be used together with --docId' + ) + }) + }) + + describe('without any stuck docs', function () { + it('should process zero docs', async function () { + const result = await runScript([ + `--estimated-docs=${await getEstimatedDocs()}`, + ]) + expect(result.code).to.equal(0) + expect(result.stdout).to.include( + 'Processed 0 docs with pending updates: flushed=0 missing=0 discarded=0 errored=0' + ) + }) + }) +})