|
|
@@ -0,0 +1,324 @@
|
|
|
+// @ts-check
|
|
|
+import Events from 'node:events'
|
|
|
+import { setTimeout } from 'node:timers/promises'
|
|
|
+import readline from 'node:readline'
|
|
|
+import fs from 'node:fs'
|
|
|
+import minimist from 'minimist'
|
|
|
+import { ObjectId } from 'mongodb'
|
|
|
+import { batchedUpdate } from '@overleaf/mongo-utils/batchedUpdate.js'
|
|
|
+import logger from '@overleaf/logger'
|
|
|
+import Metrics from '@overleaf/metrics'
|
|
|
+import OError from '@overleaf/o-error'
|
|
|
+import { promiseMapWithLimit } from '@overleaf/promise-utils'
|
|
|
+import { db, mongoClient } from '../app/js/mongodb.js'
|
|
|
+import * as HistoryStoreManager from '../app/js/HistoryStoreManager.js'
|
|
|
+import * as RedisManager from '../app/js/RedisManager.js'
|
|
|
+import * as SyncManager from '../app/js/SyncManager.js'
|
|
|
+import * as UpdatesProcessor from '../app/js/UpdatesProcessor.js'
|
|
|
+import { NeedFullProjectStructureResyncError } from '../app/js/Errors.js'
|
|
|
+import * as ErrorRecorder from '../app/js/ErrorRecorder.js'
|
|
|
+
|
|
|
+// Silence warning.
|
|
|
+Events.setMaxListeners(20)
|
|
|
+
|
|
|
+// Enable caching for ObjectId.toString()
|
|
|
+ObjectId.cacheHexString = true
|
|
|
+
|
|
|
+const READ_CONCURRENCY = parseInt(process.env.READ_CONCURRENCY || '100', 10)
|
|
|
+const WRITE_CONCURRENCY = parseInt(process.env.WRITE_CONCURRENCY || '10', 10)
|
|
|
+const FLUSH_RETRIES = parseInt(process.env.FLUSH_RETRIES || '20', 10)
|
|
|
+
|
|
|
+// Relevant dates:
|
|
|
+// - 2024-12-19, start of event-hold removal in filestore bucket -> objects older than 24h are (soft-)deleted.
|
|
|
+// - 2024-12-23, copy operation skipped in filestore when cloning project -> objects not created on clone.
|
|
|
+// - 2025-01-24, no more filestore reads allowed in project-history -> no more empty files in history for 404s
|
|
|
+const FILESTORE_SOFT_DELETE_START = new Date('2024-12-19T00:00:00Z')
|
|
|
+const FILESTORE_READ_OFF = new Date('2025-01-24T15:00:00Z')
|
|
|
+
|
|
|
+const argv = minimist(process.argv.slice(2), {
|
|
|
+ string: ['logs'],
|
|
|
+})
|
|
|
+
|
|
|
+let gracefulShutdownInitiated = false
|
|
|
+
|
|
|
+process.on('SIGINT', handleSignal)
|
|
|
+process.on('SIGTERM', handleSignal)
|
|
|
+
|
|
|
+function handleSignal() {
|
|
|
+ gracefulShutdownInitiated = true
|
|
|
+ console.warn('graceful shutdown initiated, draining queue')
|
|
|
+}
|
|
|
+
|
|
|
+const STATS = {
|
|
|
+ processedLines: 0,
|
|
|
+ success: 0,
|
|
|
+ changed: 0,
|
|
|
+ failure: 0,
|
|
|
+ skipped: 0,
|
|
|
+ checkFailure: 0,
|
|
|
+}
|
|
|
+
|
|
|
+function logStats() {
|
|
|
+ console.log(
|
|
|
+ JSON.stringify({
|
|
|
+ time: new Date(),
|
|
|
+ gracefulShutdownInitiated,
|
|
|
+ ...STATS,
|
|
|
+ })
|
|
|
+ )
|
|
|
+}
|
|
|
+const logInterval = setInterval(logStats, 10_000)
|
|
|
+
|
|
|
+/**
|
|
|
+ * @typedef {Object} FileRef
|
|
|
+ * @property {ObjectId} _id
|
|
|
+ * @property {any} linkedFileData
|
|
|
+ */
|
|
|
+
|
|
|
+/**
|
|
|
+ * @typedef {Object} Folder
|
|
|
+ * @property {Array<Folder>} folders
|
|
|
+ * @property {Array<FileRef>} fileRefs
|
|
|
+ */
|
|
|
+
|
|
|
+/**
|
|
|
+ * @typedef {Object} Project
|
|
|
+ * @property {ObjectId} _id
|
|
|
+ * @property {Date} lastUpdated
|
|
|
+ * @property {Array<Folder>} rootFolder
|
|
|
+ * @property {{history: {id: (number|string)}}} overleaf
|
|
|
+ */
|
|
|
+
|
|
|
+/**
|
|
|
+ * @param {Folder} folder
|
|
|
+ * @return {boolean}
|
|
|
+ */
|
|
|
+function checkFileTreeNeedsResync(folder) {
|
|
|
+ if (!folder) return false
|
|
|
+ if (Array.isArray(folder.fileRefs)) {
|
|
|
+ for (const fileRef of folder.fileRefs) {
|
|
|
+ if (fileRef.linkedFileData) return true
|
|
|
+ if (fileRef._id.getTimestamp() > FILESTORE_SOFT_DELETE_START) return true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (Array.isArray(folder.folders)) {
|
|
|
+ for (const child of folder.folders) {
|
|
|
+ if (checkFileTreeNeedsResync(child)) return true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * @param {string} projectId
|
|
|
+ * @param {string} historyId
|
|
|
+ * @return {Promise<Date>}
|
|
|
+ */
|
|
|
+async function getLastEndTimestamp(projectId, historyId) {
|
|
|
+ const raw = await HistoryStoreManager.promises.getMostRecentVersionRaw(
|
|
|
+ projectId,
|
|
|
+ historyId
|
|
|
+ )
|
|
|
+ if (!raw) throw new Error('bug: history not initialized')
|
|
|
+ return raw.endTimestamp
|
|
|
+}
|
|
|
+
|
|
|
+/** @type {Record<string, (project: Project) => Promise<boolean>>} */
|
|
|
+const conditions = {
|
|
|
+ // cheap: in-memory mongo lookup
|
|
|
+ 'updated after filestore soft-delete': async function (project) {
|
|
|
+ return project.lastUpdated > FILESTORE_SOFT_DELETE_START
|
|
|
+ },
|
|
|
+ // cheap: in-memory mongo lookup
|
|
|
+ 'file-tree requires re-sync': async function (project) {
|
|
|
+ return checkFileTreeNeedsResync(project.rootFolder?.[0])
|
|
|
+ },
|
|
|
+ // moderate: GET from Redis
|
|
|
+ 'has pending operations': async function (project) {
|
|
|
+ const n = await RedisManager.promises.countUnprocessedUpdates(
|
|
|
+ project._id.toString()
|
|
|
+ )
|
|
|
+ return n > 0
|
|
|
+ },
|
|
|
+ // expensive: GET from Mongo/Postgres via history-v1 HTTP API call
|
|
|
+ 'has been flushed after filestore soft-delete': async function (project) {
|
|
|
+ // Resyncs started after soft-deleting can trigger 404s and result in empty files.
|
|
|
+ const endTimestamp = await getLastEndTimestamp(
|
|
|
+ project._id.toString(),
|
|
|
+ project.overleaf.history.id.toString()
|
|
|
+ )
|
|
|
+ return endTimestamp > FILESTORE_SOFT_DELETE_START
|
|
|
+ },
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * @param {Project} project
|
|
|
+ * @return {Promise<{projectId: string, historyId: string} | null>}
|
|
|
+ */
|
|
|
+async function checkProject(project) {
|
|
|
+ if (gracefulShutdownInitiated) return null
|
|
|
+ if (project._id.getTimestamp() > FILESTORE_READ_OFF) {
|
|
|
+ STATS.skipped++ // Project created after all bugs were fixed.
|
|
|
+ return null
|
|
|
+ }
|
|
|
+ const projectId = project._id.toString()
|
|
|
+ const historyId = project.overleaf.history.id.toString()
|
|
|
+ for (const [condition, check] of Object.entries(conditions)) {
|
|
|
+ try {
|
|
|
+ if (await check(project)) return { projectId, historyId }
|
|
|
+ } catch (err) {
|
|
|
+ logger.err({ projectId, condition, err }, 'failed to check project')
|
|
|
+ STATS.checkFailure++
|
|
|
+ return null
|
|
|
+ }
|
|
|
+ }
|
|
|
+ STATS.skipped++
|
|
|
+ return null
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * @param {string} projectId
|
|
|
+ * @param {string} historyId
|
|
|
+ * @return {Promise<void>}
|
|
|
+ */
|
|
|
+async function processProject(projectId, historyId) {
|
|
|
+ if (gracefulShutdownInitiated) return
|
|
|
+ const t0 = performance.now()
|
|
|
+ try {
|
|
|
+ await tryProcessProject(projectId, historyId)
|
|
|
+ const latency = performance.now() - t0
|
|
|
+ logger.info({ projectId, historyId, latency }, 'processed project')
|
|
|
+ STATS.success++
|
|
|
+ } catch (err) {
|
|
|
+ logger.err({ err, projectId, historyId }, 'failed to process project')
|
|
|
+ STATS.failure++
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * @param {string} projectId
|
|
|
+ * @return {Promise<void>}
|
|
|
+ */
|
|
|
+async function flushWithRetries(projectId) {
|
|
|
+ for (let attempt = 0; attempt < FLUSH_RETRIES; attempt++) {
|
|
|
+ try {
|
|
|
+ await UpdatesProcessor.promises.processUpdatesForProject(projectId)
|
|
|
+ return
|
|
|
+ } catch (err) {
|
|
|
+ logger.warn(
|
|
|
+ { projectId, err, attempt },
|
|
|
+ 'failed to flush updates, trying again'
|
|
|
+ )
|
|
|
+ if (gracefulShutdownInitiated) throw err
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ await UpdatesProcessor.promises.processUpdatesForProject(projectId)
|
|
|
+ } catch (err) {
|
|
|
+ // @ts-ignore err is Error
|
|
|
+ throw new OError('failed to flush updates', {}, err)
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * @param {string} projectId
|
|
|
+ * @param {string} historyId
|
|
|
+ * @return {Promise<void>}
|
|
|
+ */
|
|
|
+async function tryProcessProject(projectId, historyId) {
|
|
|
+ await flushWithRetries(projectId)
|
|
|
+ const start = new Date()
|
|
|
+ let needsFullSync = false
|
|
|
+ try {
|
|
|
+ await UpdatesProcessor.promises.startResyncAndProcessUpdatesUnderLock(
|
|
|
+ projectId,
|
|
|
+ { resyncProjectStructureOnly: true }
|
|
|
+ )
|
|
|
+ } catch (err) {
|
|
|
+ if (err instanceof NeedFullProjectStructureResyncError) {
|
|
|
+ needsFullSync = true
|
|
|
+ } else {
|
|
|
+ throw err
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (needsFullSync) {
|
|
|
+ logger.warn(
|
|
|
+ { projectId, historyId },
|
|
|
+ 'structure only resync not sufficient, doing full soft resync'
|
|
|
+ )
|
|
|
+ await SyncManager.promises.startResync(projectId, {})
|
|
|
+ await UpdatesProcessor.promises.processUpdatesForProject(projectId)
|
|
|
+ STATS.changed++
|
|
|
+ } else {
|
|
|
+ const after = await getLastEndTimestamp(projectId, historyId)
|
|
|
+ if (after > start) {
|
|
|
+ STATS.changed++
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // Avoid db.projectHistorySyncState from growing for each project we resynced.
|
|
|
+ // MongoDB collections cannot shrink on their own. In case of success, purge
|
|
|
+ // the db entry created by this script right away.
|
|
|
+ await SyncManager.promises.clearResyncStateIfAllAfter(projectId, start)
|
|
|
+}
|
|
|
+
|
|
|
+async function processBatch(projects) {
|
|
|
+ const projectIds = (
|
|
|
+ await promiseMapWithLimit(READ_CONCURRENCY, projects, checkProject)
|
|
|
+ ).filter(id => !!id)
|
|
|
+ await promiseMapWithLimit(WRITE_CONCURRENCY, projectIds, ids =>
|
|
|
+ processProject(ids.projectId, ids.historyId)
|
|
|
+ )
|
|
|
+
|
|
|
+ if (gracefulShutdownInitiated) throw new Error('graceful shutdown triggered')
|
|
|
+}
|
|
|
+
|
|
|
+async function processProjectsFromLog() {
|
|
|
+ const rl = readline.createInterface({
|
|
|
+ input: fs.createReadStream(argv.logs),
|
|
|
+ })
|
|
|
+ for await (const line of rl) {
|
|
|
+ if (gracefulShutdownInitiated) break
|
|
|
+ STATS.processedLines++
|
|
|
+ if (!line.startsWith('{')) continue
|
|
|
+ const { projectId, historyId, msg } = JSON.parse(line)
|
|
|
+ if (msg !== 'failed to process project') continue
|
|
|
+ await processProject(projectId, historyId) // does try/catch with logging
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+async function main() {
|
|
|
+ if (argv.logs) {
|
|
|
+ await processProjectsFromLog()
|
|
|
+ return
|
|
|
+ }
|
|
|
+ await batchedUpdate(db.projects, {}, processBatch, {
|
|
|
+ _id: 1,
|
|
|
+ lastUpdated: 1,
|
|
|
+ 'overleaf.history': 1,
|
|
|
+ rootFolder: 1,
|
|
|
+ })
|
|
|
+}
|
|
|
+
|
|
|
+try {
|
|
|
+ try {
|
|
|
+ await main()
|
|
|
+ } finally {
|
|
|
+ clearInterval(logInterval)
|
|
|
+ logStats()
|
|
|
+ Metrics.close()
|
|
|
+ await mongoClient.close()
|
|
|
+ // TODO(das7pad): graceful shutdown for redis. Refactor process.exit when done.
|
|
|
+ }
|
|
|
+ console.log('Done.')
|
|
|
+ await setTimeout(1_000)
|
|
|
+ if (STATS.failure) {
|
|
|
+ process.exit(Math.min(STATS.failure, 99))
|
|
|
+ } else {
|
|
|
+ process.exit(0)
|
|
|
+ }
|
|
|
+} catch (err) {
|
|
|
+ logger.err({ err }, 'fatal error')
|
|
|
+ await setTimeout(1_000)
|
|
|
+ process.exit(100)
|
|
|
+}
|
Jakob Ackermann