backup.mjs 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038
  1. // @ts-check
  2. import logger from '@overleaf/logger'
  3. import commandLineArgs from 'command-line-args'
  4. import { History } from 'overleaf-editor-core'
  5. import { getProjectChunks, loadLatestRaw } from '../lib/chunk_store/index.js'
  6. import { client } from '../lib/mongodb.js'
  7. import knex from '../lib/knex.js'
  8. import { historyStore } from '../lib/history_store.js'
  9. import pLimit from 'p-limit'
  10. import {
  11. GLOBAL_BLOBS,
  12. loadGlobalBlobs,
  13. makeProjectKey,
  14. BlobStore,
  15. } from '../lib/blob_store/index.js'
  16. import {
  17. listPendingBackups,
  18. getBackupStatus,
  19. setBackupVersion,
  20. updateCurrentMetadataIfNotSet,
  21. updatePendingChangeTimestamp,
  22. getBackedUpBlobHashes,
  23. unsetBackedUpBlobHashes,
  24. } from '../lib/backup_store/index.js'
  25. import { backupBlob, downloadBlobToDir } from '../lib/backupBlob.mjs'
  26. import {
  27. backupPersistor,
  28. chunksBucket,
  29. projectBlobsBucket,
  30. } from '../lib/backupPersistor.mjs'
  31. import { backupGenerator } from '../lib/backupGenerator.mjs'
  32. import { promises as fs } from 'node:fs'
  33. import os from 'node:os'
  34. import path from 'node:path'
  35. import projectKey from '../lib/project_key.js'
  36. import Crypto from 'node:crypto'
  37. import Stream from 'node:stream'
  38. import { EventEmitter } from 'node:events'
  39. import { batchedUpdate } from '@overleaf/mongo-utils/batchedUpdate.js'
  40. import { createGunzip } from 'node:zlib'
  41. import { text } from 'node:stream/consumers'
  42. import { fromStream as blobHashFromStream } from '../lib/blob_hash.js'
  43. import { NotFoundError } from '@overleaf/object-persistor/src/Errors.js'
  44. // Create a singleton promise that loads global blobs once
  45. let globalBlobsPromise = null
  46. function ensureGlobalBlobsLoaded() {
  47. if (!globalBlobsPromise) {
  48. globalBlobsPromise = loadGlobalBlobs()
  49. }
  50. return globalBlobsPromise
  51. }
  52. EventEmitter.defaultMaxListeners = 20
  53. logger.initialize('history-v1-backup')
  54. // Settings shared between command-line and module usage
  55. let DRY_RUN = false
  56. let RETRY_LIMIT = 3
  57. const RETRY_DELAY = 1000
  58. let CONCURRENCY = 4
  59. let BATCH_CONCURRENCY = 1
  60. let BLOB_LIMITER = pLimit(CONCURRENCY)
  61. let USE_SECONDARY = false
  62. /**
  63. * Configure backup settings
  64. * @param {Object} options Backup configuration options
  65. */
  66. export function configureBackup(options = {}) {
  67. DRY_RUN = options.dryRun || false
  68. RETRY_LIMIT = options.retries || 3
  69. CONCURRENCY = options.concurrency || 1
  70. BATCH_CONCURRENCY = options.batchConcurrency || 1
  71. BLOB_LIMITER = pLimit(CONCURRENCY)
  72. USE_SECONDARY = options.useSecondary || false
  73. }
  74. let gracefulShutdownInitiated = false
  75. process.on('SIGINT', handleSignal)
  76. process.on('SIGTERM', handleSignal)
  77. function handleSignal() {
  78. gracefulShutdownInitiated = true
  79. console.warn('graceful shutdown initiated, draining queue')
  80. }
  81. async function retry(fn, times, delayMs) {
  82. let attempts = times
  83. while (attempts > 0) {
  84. try {
  85. const result = await fn()
  86. return result
  87. } catch (err) {
  88. attempts--
  89. if (attempts === 0) throw err
  90. await new Promise(resolve => setTimeout(resolve, delayMs))
  91. }
  92. }
  93. }
  94. function wrapWithRetry(fn, retries, delayMs) {
  95. return async (...args) => {
  96. const result = await retry(() => fn(...args), retries, delayMs)
  97. return result
  98. }
  99. }
  100. const downloadWithRetry = wrapWithRetry(
  101. downloadBlobToDir,
  102. RETRY_LIMIT,
  103. RETRY_DELAY
  104. )
  105. // FIXME: this creates a new backupPersistor for each blob
  106. // so there is no caching of the DEK
  107. const backupWithRetry = wrapWithRetry(backupBlob, RETRY_LIMIT, RETRY_DELAY)
  108. async function findNewBlobs(projectId, blobs) {
  109. const newBlobs = []
  110. const existingBackedUpBlobHashes = await getBackedUpBlobHashes(projectId)
  111. for (const blob of blobs) {
  112. const hash = blob.getHash()
  113. if (existingBackedUpBlobHashes.has(blob.getHash())) {
  114. logger.debug({ projectId, hash }, 'Blob is already backed up, skipping')
  115. continue
  116. }
  117. const globalBlob = GLOBAL_BLOBS.get(hash)
  118. if (globalBlob && !globalBlob.demoted) {
  119. logger.debug(
  120. { projectId, hash },
  121. 'Blob is a global blob and not demoted, skipping'
  122. )
  123. continue
  124. }
  125. newBlobs.push(blob)
  126. }
  127. return newBlobs
  128. }
  129. async function cleanBackedUpBlobs(projectId, blobs) {
  130. const hashes = blobs.map(blob => blob.getHash())
  131. if (DRY_RUN) {
  132. console.log(
  133. 'Would remove blobs',
  134. hashes.join(' '),
  135. 'from project',
  136. projectId
  137. )
  138. return
  139. }
  140. await unsetBackedUpBlobHashes(projectId, hashes)
  141. }
  142. async function backupSingleBlob(projectId, historyId, blob, tmpDir, persistor) {
  143. if (DRY_RUN) {
  144. console.log(
  145. 'Would back up blob',
  146. JSON.stringify(blob),
  147. 'in history',
  148. historyId,
  149. 'for project',
  150. projectId
  151. )
  152. return
  153. }
  154. logger.debug({ blob, historyId }, 'backing up blob')
  155. const blobPath = await downloadWithRetry(historyId, blob, tmpDir)
  156. await backupWithRetry(historyId, blob, blobPath, persistor)
  157. }
  158. async function backupBlobs(projectId, historyId, blobs, limiter, persistor) {
  159. let tmpDir
  160. try {
  161. tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'blob-backup-'))
  162. const blobBackupOperations = blobs.map(blob =>
  163. limiter(backupSingleBlob, projectId, historyId, blob, tmpDir, persistor)
  164. )
  165. await Promise.allSettled(blobBackupOperations)
  166. } finally {
  167. if (tmpDir) {
  168. await fs.rm(tmpDir, { recursive: true, force: true })
  169. }
  170. }
  171. }
  172. async function backupChunk(
  173. projectId,
  174. historyId,
  175. chunkBackupPersistorForProject,
  176. chunkToBackup,
  177. chunkRecord,
  178. chunkBuffer
  179. ) {
  180. if (DRY_RUN) {
  181. console.log(
  182. 'Would back up chunk',
  183. JSON.stringify(chunkRecord),
  184. 'in history',
  185. historyId,
  186. 'for project',
  187. projectId,
  188. 'key',
  189. makeChunkKey(historyId, chunkToBackup.startVersion)
  190. )
  191. return
  192. }
  193. const key = makeChunkKey(historyId, chunkToBackup.startVersion)
  194. logger.debug({ chunkRecord, historyId, projectId, key }, 'backing up chunk')
  195. const md5 = Crypto.createHash('md5').update(chunkBuffer)
  196. await chunkBackupPersistorForProject.sendStream(
  197. chunksBucket,
  198. makeChunkKey(historyId, chunkToBackup.startVersion),
  199. Stream.Readable.from([chunkBuffer]),
  200. {
  201. contentType: 'application/json',
  202. contentEncoding: 'gzip',
  203. contentLength: chunkBuffer.byteLength,
  204. sourceMd5: md5.digest('hex'),
  205. }
  206. )
  207. }
  208. async function updateBackupStatus(
  209. projectId,
  210. lastBackedUpVersion,
  211. chunkRecord,
  212. startOfBackupTime
  213. ) {
  214. if (DRY_RUN) {
  215. console.log(
  216. 'Would set backup version to',
  217. chunkRecord.endVersion,
  218. 'with lastBackedUpTimestamp',
  219. startOfBackupTime
  220. )
  221. return
  222. }
  223. logger.debug(
  224. { projectId, chunkRecord, startOfBackupTime },
  225. 'setting backupVersion and lastBackedUpTimestamp'
  226. )
  227. await setBackupVersion(
  228. projectId,
  229. lastBackedUpVersion,
  230. chunkRecord.endVersion,
  231. startOfBackupTime
  232. )
  233. }
  234. // Define command-line options
  235. const optionDefinitions = [
  236. {
  237. name: 'projectId',
  238. alias: 'p',
  239. type: String,
  240. description: 'The ID of the project to backup',
  241. defaultOption: true,
  242. },
  243. {
  244. name: 'help',
  245. alias: 'h',
  246. type: Boolean,
  247. description: 'Display this usage guide.',
  248. },
  249. {
  250. name: 'status',
  251. alias: 's',
  252. type: Boolean,
  253. description: 'Display project status.',
  254. },
  255. {
  256. name: 'list',
  257. alias: 'l',
  258. type: Boolean,
  259. description: 'List projects that need to be backed up',
  260. },
  261. {
  262. name: 'dry-run',
  263. alias: 'n',
  264. type: Boolean,
  265. description: 'Perform a dry run without making any changes.',
  266. },
  267. {
  268. name: 'retries',
  269. alias: 'r',
  270. type: Number,
  271. description: 'Number of retries, default is 3.',
  272. },
  273. {
  274. name: 'concurrency',
  275. alias: 'c',
  276. type: Number,
  277. description: 'Number of concurrent blob downloads (default: 1)',
  278. },
  279. {
  280. name: 'batch-concurrency',
  281. alias: 'b',
  282. type: Number,
  283. description: 'Number of concurrent project operations (default: 1)',
  284. },
  285. {
  286. name: 'pending',
  287. alias: 'P',
  288. type: Boolean,
  289. description: 'Backup all pending projects.',
  290. },
  291. {
  292. name: 'interval',
  293. alias: 'i',
  294. type: Number,
  295. description: 'Time interval in seconds for pending backups (default: 3600)',
  296. defaultValue: 3600,
  297. },
  298. {
  299. name: 'init',
  300. alias: 'I',
  301. type: Boolean,
  302. description: 'Initialize backups for all projects.',
  303. },
  304. {
  305. name: 'start-date',
  306. type: String,
  307. description: 'Start date for initialization (ISO format)',
  308. },
  309. {
  310. name: 'end-date',
  311. type: String,
  312. description: 'End date for initialization (ISO format)',
  313. },
  314. {
  315. name: 'use-secondary',
  316. type: Boolean,
  317. description: 'Use secondary read preference for backup status',
  318. },
  319. {
  320. name: 'compare',
  321. alias: 'C',
  322. type: Boolean,
  323. description:
  324. 'Compare backup with original chunks. With --start-date and --end-date compares all projects in range.',
  325. },
  326. ]
  327. function handleOptions() {
  328. const options = commandLineArgs(optionDefinitions)
  329. if (options.help) {
  330. console.log('Usage:')
  331. optionDefinitions.forEach(option => {
  332. console.log(` --${option.name}, -${option.alias}: ${option.description}`)
  333. })
  334. process.exit(0)
  335. }
  336. const projectIdRequired =
  337. !options.list &&
  338. !options.pending &&
  339. !options.init &&
  340. !(options.compare && options['start-date'] && options['end-date'])
  341. if (projectIdRequired && !options.projectId) {
  342. console.error('Error: projectId is required')
  343. process.exit(1)
  344. }
  345. if (options.pending && options.projectId) {
  346. console.error('Error: --pending cannot be specified with projectId')
  347. process.exit(1)
  348. }
  349. if (options.pending && (options.list || options.status)) {
  350. console.error('Error: --pending is exclusive with --list and --status')
  351. process.exit(1)
  352. }
  353. if (options.init && options.pending) {
  354. console.error('Error: --init cannot be specified with --pending')
  355. process.exit(1)
  356. }
  357. if (
  358. (options['start-date'] || options['end-date']) &&
  359. !options.init &&
  360. !options.compare
  361. ) {
  362. console.error(
  363. 'Error: date options can only be used with --init or --compare'
  364. )
  365. process.exit(1)
  366. }
  367. if (options['use-secondary']) {
  368. USE_SECONDARY = true
  369. }
  370. if (
  371. options.compare &&
  372. !options.projectId &&
  373. !(options['start-date'] && options['end-date'])
  374. ) {
  375. console.error(
  376. 'Error: --compare requires either projectId or both --start-date and --end-date'
  377. )
  378. process.exit(1)
  379. }
  380. DRY_RUN = options['dry-run'] || false
  381. RETRY_LIMIT = options.retries || 3
  382. CONCURRENCY = options.concurrency || 1
  383. BATCH_CONCURRENCY = options['batch-concurrency'] || 1
  384. BLOB_LIMITER = pLimit(CONCURRENCY)
  385. return options
  386. }
  387. async function displayBackupStatus(projectId) {
  388. const result = await analyseBackupStatus(projectId)
  389. console.log('Backup status:', JSON.stringify(result))
  390. }
  391. async function analyseBackupStatus(projectId) {
  392. const { backupStatus, historyId, currentEndVersion, currentEndTimestamp } =
  393. await getBackupStatus(projectId)
  394. // TODO: when we have confidence that the latestChunkMetadata always matches
  395. // the values from the backupStatus we can skip loading it here
  396. const latestChunkMetadata = await loadLatestRaw(historyId, {
  397. readOnly: Boolean(USE_SECONDARY),
  398. })
  399. if (
  400. currentEndVersion &&
  401. currentEndVersion !== latestChunkMetadata.endVersion
  402. ) {
  403. // compare the current end version with the latest chunk metadata to check that
  404. // the updates to the project collection are reliable
  405. // expect some failures due to the time window between getBackupStatus and
  406. // loadLatestRaw where the project is being actively edited.
  407. logger.warn(
  408. {
  409. projectId,
  410. historyId,
  411. currentEndVersion,
  412. currentEndTimestamp,
  413. latestChunkMetadata,
  414. },
  415. 'currentEndVersion does not match latest chunk metadata'
  416. )
  417. }
  418. if (DRY_RUN) {
  419. console.log('Project:', projectId)
  420. console.log('History ID:', historyId)
  421. console.log('Latest Chunk Metadata:', JSON.stringify(latestChunkMetadata))
  422. console.log('Current end version:', currentEndVersion)
  423. console.log('Current end timestamp:', currentEndTimestamp)
  424. console.log('Backup status:', backupStatus ?? 'none')
  425. }
  426. if (!backupStatus) {
  427. if (DRY_RUN) {
  428. console.log('No backup status found - doing full backup')
  429. }
  430. }
  431. const lastBackedUpVersion = backupStatus?.lastBackedUpVersion
  432. const endVersion = latestChunkMetadata.endVersion
  433. if (endVersion >= 0 && endVersion === lastBackedUpVersion) {
  434. if (DRY_RUN) {
  435. console.log(
  436. 'Project is up to date, last backed up at version',
  437. lastBackedUpVersion
  438. )
  439. }
  440. } else if (endVersion < lastBackedUpVersion) {
  441. throw new Error('backup is ahead of project')
  442. } else {
  443. if (DRY_RUN) {
  444. console.log(
  445. 'Project needs to be backed up from',
  446. lastBackedUpVersion,
  447. 'to',
  448. endVersion
  449. )
  450. }
  451. }
  452. return {
  453. historyId,
  454. lastBackedUpVersion,
  455. currentVersion: latestChunkMetadata.endVersion || 0,
  456. upToDate: endVersion >= 0 && lastBackedUpVersion === endVersion,
  457. pendingChangeAt: backupStatus?.pendingChangeAt,
  458. currentEndVersion,
  459. currentEndTimestamp,
  460. latestChunkMetadata,
  461. }
  462. }
  463. async function displayPendingBackups(options) {
  464. const intervalMs = options.interval * 1000
  465. for await (const project of listPendingBackups(intervalMs)) {
  466. console.log(
  467. 'Project:',
  468. project._id.toHexString(),
  469. 'backup status:',
  470. JSON.stringify(project.overleaf.backup),
  471. 'history status:',
  472. JSON.stringify(project.overleaf.history, [
  473. 'currentEndVersion',
  474. 'currentEndTimestamp',
  475. ])
  476. )
  477. }
  478. }
  479. function makeChunkKey(projectId, startVersion) {
  480. return path.join(projectKey.format(projectId), projectKey.pad(startVersion))
  481. }
  482. export async function backupProject(projectId, options) {
  483. if (gracefulShutdownInitiated) {
  484. return
  485. }
  486. await ensureGlobalBlobsLoaded()
  487. // FIXME: flush the project first!
  488. // Let's assume the the flush happens externally and triggers this backup
  489. const backupStartTime = new Date()
  490. // find the last backed up version
  491. const {
  492. historyId,
  493. lastBackedUpVersion,
  494. currentVersion,
  495. upToDate,
  496. pendingChangeAt,
  497. currentEndVersion,
  498. latestChunkMetadata,
  499. } = await analyseBackupStatus(projectId)
  500. if (upToDate) {
  501. logger.debug(
  502. {
  503. projectId,
  504. historyId,
  505. lastBackedUpVersion,
  506. currentVersion,
  507. pendingChangeAt,
  508. },
  509. 'backup is up to date'
  510. )
  511. if (
  512. currentEndVersion === undefined &&
  513. latestChunkMetadata.endVersion >= 0
  514. ) {
  515. if (DRY_RUN) {
  516. console.log('Would update current metadata to', latestChunkMetadata)
  517. } else {
  518. await updateCurrentMetadataIfNotSet(projectId, latestChunkMetadata)
  519. }
  520. }
  521. // clear the pending changes timestamp if the backup is complete
  522. if (pendingChangeAt) {
  523. if (DRY_RUN) {
  524. console.log(
  525. 'Would update or clear pending changes timestamp',
  526. backupStartTime
  527. )
  528. } else {
  529. await updatePendingChangeTimestamp(projectId, backupStartTime)
  530. }
  531. }
  532. return
  533. }
  534. logger.debug(
  535. {
  536. projectId,
  537. historyId,
  538. lastBackedUpVersion,
  539. currentVersion,
  540. pendingChangeAt,
  541. },
  542. 'backing up project'
  543. )
  544. // this persistor works for both the chunks and blobs buckets,
  545. // because they use the same DEK
  546. const backupPersistorForProject = await backupPersistor.forProject(
  547. chunksBucket,
  548. makeProjectKey(historyId, '')
  549. )
  550. let previousBackedUpVersion = lastBackedUpVersion
  551. const backupVersions = [previousBackedUpVersion]
  552. for await (const {
  553. blobsToBackup,
  554. chunkToBackup,
  555. chunkRecord,
  556. chunkBuffer,
  557. } of backupGenerator(historyId, lastBackedUpVersion)) {
  558. // backup the blobs first
  559. // this can be done in parallel but must fail if any blob cannot be backed up
  560. // if the blob already exists in the backup then that is allowed
  561. const newBlobs = await findNewBlobs(projectId, blobsToBackup)
  562. await backupBlobs(
  563. projectId,
  564. historyId,
  565. newBlobs,
  566. BLOB_LIMITER,
  567. backupPersistorForProject
  568. )
  569. // then backup the original compressed chunk using the startVersion as the key
  570. await backupChunk(
  571. projectId,
  572. historyId,
  573. backupPersistorForProject,
  574. chunkToBackup,
  575. chunkRecord,
  576. chunkBuffer
  577. )
  578. // persist the backup status in mongo for the current chunk
  579. try {
  580. await updateBackupStatus(
  581. projectId,
  582. previousBackedUpVersion,
  583. chunkRecord,
  584. backupStartTime
  585. )
  586. } catch (err) {
  587. logger.error(
  588. { projectId, chunkRecord, err, backupVersions },
  589. 'error updating backup status'
  590. )
  591. throw err
  592. }
  593. previousBackedUpVersion = chunkRecord.endVersion
  594. backupVersions.push(previousBackedUpVersion)
  595. await cleanBackedUpBlobs(projectId, blobsToBackup)
  596. }
  597. // update the current end version and timestamp if they are not set
  598. if (currentEndVersion === undefined && latestChunkMetadata.endVersion >= 0) {
  599. if (DRY_RUN) {
  600. console.log('Would update current metadata to', latestChunkMetadata)
  601. } else {
  602. await updateCurrentMetadataIfNotSet(projectId, latestChunkMetadata)
  603. }
  604. }
  605. // clear the pending changes timestamp if the backup is complete, otherwise set it to the time
  606. // when the backup started (to pick up the new changes on the next backup)
  607. if (DRY_RUN) {
  608. console.log(
  609. 'Would update or clear pending changes timestamp',
  610. backupStartTime
  611. )
  612. } else {
  613. await updatePendingChangeTimestamp(projectId, backupStartTime)
  614. }
  615. }
  616. function convertToISODate(dateStr) {
  617. if (!dateStr) return undefined
  618. // Expecting YYYY-MM-DD format
  619. if (!/^\d{4}-\d{2}-\d{2}$/.test(dateStr)) {
  620. throw new Error('Date must be in YYYY-MM-DD format')
  621. }
  622. return new Date(dateStr + 'T00:00:00.000Z').toISOString()
  623. }
  624. export async function initializeProjects(options) {
  625. await ensureGlobalBlobsLoaded()
  626. const limiter = pLimit(BATCH_CONCURRENCY)
  627. let totalErrors = 0
  628. let totalProjects = 0
  629. async function backupProjectWithErrorLogging(projectId) {
  630. try {
  631. await backupProject(projectId, options)
  632. } catch (err) {
  633. logger.error({ projectId, err }, 'error backing up project')
  634. throw err
  635. }
  636. }
  637. async function processBatch(batch) {
  638. if (gracefulShutdownInitiated) {
  639. throw new Error('graceful shutdown')
  640. }
  641. const batchOperations = batch.map(project =>
  642. limiter(backupProjectWithErrorLogging, project._id.toHexString())
  643. )
  644. const results = await Promise.allSettled(batchOperations)
  645. const errors = results.filter(result => result.status === 'rejected').length
  646. if (errors > 0) {
  647. logger.error(
  648. {
  649. errors,
  650. batchSize: batch.length,
  651. batchStart: batch[0]._id.toHexString(),
  652. batchEnd: batch[batch.length - 1]._id.toHexString(),
  653. },
  654. 'errors in batch'
  655. )
  656. }
  657. totalErrors += errors
  658. totalProjects += batch.length
  659. }
  660. const query = {
  661. 'overleaf.history.id': { $exists: true },
  662. 'overleaf.backup.lastBackedUpVersion': { $exists: false },
  663. 'overleaf.backup.pendingChangeAt': { $exists: false },
  664. }
  665. await batchedUpdate(
  666. client.db().collection('projects'),
  667. query,
  668. processBatch,
  669. {
  670. _id: 1,
  671. },
  672. { readPreference: 'secondary' },
  673. {
  674. BATCH_RANGE_START: convertToISODate(options['start-date']),
  675. BATCH_RANGE_END: convertToISODate(options['end-date']),
  676. }
  677. )
  678. return { errors: totalErrors, projects: totalProjects }
  679. }
  680. async function backupPendingProjects(options) {
  681. const intervalMs = options.interval * 1000
  682. for await (const project of listPendingBackups(intervalMs)) {
  683. if (gracefulShutdownInitiated) {
  684. console.warn('graceful shutdown: stopping pending project backups')
  685. break
  686. }
  687. const projectId = project._id.toHexString()
  688. console.log(`Backing up pending project with ID: ${projectId}`)
  689. await backupProject(projectId, options)
  690. }
  691. }
  692. class BlobComparator {
  693. constructor(backupPersistorForProject) {
  694. this.cache = new Map()
  695. this.backupPersistorForProject = backupPersistorForProject
  696. }
  697. async compareBlob(historyId, blob) {
  698. let computedHash = this.cache.get(blob.hash)
  699. const fromCache = !!computedHash
  700. if (!computedHash) {
  701. const blobKey = makeProjectKey(historyId, blob.hash)
  702. const backupBlobStream =
  703. await this.backupPersistorForProject.getObjectStream(
  704. projectBlobsBucket,
  705. blobKey,
  706. { autoGunzip: true }
  707. )
  708. computedHash = await blobHashFromStream(blob.byteLength, backupBlobStream)
  709. this.cache.set(blob.hash, computedHash)
  710. }
  711. const matches = computedHash === blob.hash
  712. return {
  713. matches,
  714. computedHash,
  715. fromCache,
  716. }
  717. }
  718. }
  719. async function compareBackups(projectId, options) {
  720. console.log(`Comparing backups for project ${projectId}`)
  721. const { historyId } = await getBackupStatus(projectId)
  722. const chunks = await getProjectChunks(historyId)
  723. const blobStore = new BlobStore(historyId)
  724. const backupPersistorForProject = await backupPersistor.forProject(
  725. chunksBucket,
  726. makeProjectKey(historyId, '')
  727. )
  728. let totalChunkMatches = 0
  729. let totalChunkMismatches = 0
  730. let totalChunksNotFound = 0
  731. let totalBlobMatches = 0
  732. let totalBlobMismatches = 0
  733. let totalBlobsNotFound = 0
  734. const errors = []
  735. const blobComparator = new BlobComparator(backupPersistorForProject)
  736. for (const chunk of chunks) {
  737. try {
  738. // Compare chunk content
  739. const originalChunk = await historyStore.loadRaw(historyId, chunk.id)
  740. const key = makeChunkKey(historyId, chunk.startVersion)
  741. try {
  742. const backupChunkStream =
  743. await backupPersistorForProject.getObjectStream(chunksBucket, key)
  744. const backupStr = await text(backupChunkStream.pipe(createGunzip()))
  745. const originalStr = JSON.stringify(originalChunk)
  746. const backupChunk = JSON.parse(backupStr)
  747. const backupStartVersion = chunk.startVersion
  748. const backupEndVersion = chunk.startVersion + backupChunk.changes.length
  749. if (originalStr === backupStr) {
  750. console.log(
  751. `✓ Chunk ${chunk.id} (v${chunk.startVersion}-v${chunk.endVersion}) matches`
  752. )
  753. totalChunkMatches++
  754. } else if (originalStr === JSON.stringify(JSON.parse(backupStr))) {
  755. console.log(
  756. `✓ Chunk ${chunk.id} (v${chunk.startVersion}-v${chunk.endVersion}) matches (after normalisation)`
  757. )
  758. totalChunkMatches++
  759. } else if (backupEndVersion < chunk.endVersion) {
  760. console.log(
  761. `✗ Chunk ${chunk.id} is ahead of backup (v${chunk.startVersion}-v${chunk.endVersion} vs v${backupStartVersion}-v${backupEndVersion})`
  762. )
  763. totalChunkMismatches++
  764. errors.push({ chunkId: chunk.id, error: 'Chunk ahead of backup' })
  765. } else {
  766. console.log(
  767. `✗ Chunk ${chunk.id} (v${chunk.startVersion}-v${chunk.endVersion}) MISMATCH`
  768. )
  769. totalChunkMismatches++
  770. errors.push({ chunkId: chunk.id, error: 'Chunk mismatch' })
  771. }
  772. } catch (err) {
  773. if (err instanceof NotFoundError) {
  774. console.log(`✗ Chunk ${chunk.id} not found in backup`, err.cause)
  775. totalChunksNotFound++
  776. errors.push({ chunkId: chunk.id, error: `Chunk not found` })
  777. } else {
  778. throw err
  779. }
  780. }
  781. const history = History.fromRaw(originalChunk)
  782. // Compare blobs in chunk
  783. const blobHashes = new Set()
  784. history.findBlobHashes(blobHashes)
  785. const blobs = await blobStore.getBlobs(Array.from(blobHashes))
  786. for (const blob of blobs) {
  787. if (GLOBAL_BLOBS.has(blob.hash)) {
  788. const globalBlob = GLOBAL_BLOBS.get(blob.hash)
  789. console.log(
  790. ` ✓ Blob ${blob.hash} is a global blob`,
  791. globalBlob.demoted ? '(demoted)' : ''
  792. )
  793. continue
  794. }
  795. try {
  796. const { matches, computedHash, fromCache } =
  797. await blobComparator.compareBlob(historyId, blob)
  798. if (matches) {
  799. console.log(
  800. ` ✓ Blob ${blob.hash} hash matches (${blob.byteLength} bytes)` +
  801. (fromCache ? ' (from cache)' : '')
  802. )
  803. totalBlobMatches++
  804. } else {
  805. console.log(
  806. ` ✗ Blob ${blob.hash} hash mismatch (original: ${blob.hash}, backup: ${computedHash}) (${blob.byteLength} bytes, ${blob.stringLength} string length)` +
  807. (fromCache ? ' (from cache)' : '')
  808. )
  809. totalBlobMismatches++
  810. errors.push({
  811. chunkId: chunk.id,
  812. error: `Blob ${blob.hash} hash mismatch`,
  813. })
  814. }
  815. } catch (err) {
  816. if (err instanceof NotFoundError) {
  817. console.log(` ✗ Blob ${blob.hash} not found in backup`, err.cause)
  818. totalBlobsNotFound++
  819. errors.push({
  820. chunkId: chunk.id,
  821. error: `Blob ${blob.hash} not found`,
  822. })
  823. } else {
  824. throw err
  825. }
  826. }
  827. }
  828. } catch (err) {
  829. console.error(`Error comparing chunk ${chunk.id}:`, err)
  830. errors.push({ chunkId: chunk.id, error: err })
  831. }
  832. }
  833. // Print summary
  834. console.log('\nComparison Summary:')
  835. console.log('==================')
  836. console.log(`Total chunks: ${chunks.length}`)
  837. console.log(`Chunk matches: ${totalChunkMatches}`)
  838. console.log(`Chunk mismatches: ${totalChunkMismatches}`)
  839. console.log(`Chunk not found: ${totalChunksNotFound}`)
  840. console.log(`Blob matches: ${totalBlobMatches}`)
  841. console.log(`Blob mismatches: ${totalBlobMismatches}`)
  842. console.log(`Blob not found: ${totalBlobsNotFound}`)
  843. console.log(`Errors: ${errors.length}`)
  844. if (errors.length > 0) {
  845. console.log('\nErrors:')
  846. errors.forEach(({ chunkId, error }) => {
  847. console.log(` Chunk ${chunkId}: ${error}`)
  848. })
  849. throw new Error('Backup comparison FAILED')
  850. } else {
  851. console.log('Backup comparison successful')
  852. }
  853. }
  854. async function compareAllProjects(options) {
  855. const limiter = pLimit(BATCH_CONCURRENCY)
  856. let totalErrors = 0
  857. let totalProjects = 0
  858. async function processBatch(batch) {
  859. if (gracefulShutdownInitiated) {
  860. throw new Error('graceful shutdown')
  861. }
  862. const batchOperations = batch.map(project =>
  863. limiter(async () => {
  864. const projectId = project._id.toHexString()
  865. totalProjects++
  866. try {
  867. console.log(`\nComparing project ${projectId} (${totalProjects})`)
  868. await compareBackups(projectId, options)
  869. } catch (err) {
  870. totalErrors++
  871. console.error(`Failed to compare project ${projectId}:`, err)
  872. }
  873. })
  874. )
  875. await Promise.allSettled(batchOperations)
  876. }
  877. const query = {
  878. 'overleaf.history.id': { $exists: true },
  879. 'overleaf.backup.lastBackedUpVersion': { $exists: true },
  880. }
  881. await batchedUpdate(
  882. client.db().collection('projects'),
  883. query,
  884. processBatch,
  885. {
  886. _id: 1,
  887. 'overleaf.history': 1,
  888. 'overleaf.backup': 1,
  889. },
  890. { readPreference: 'secondary' },
  891. {
  892. BATCH_RANGE_START: convertToISODate(options['start-date']),
  893. BATCH_RANGE_END: convertToISODate(options['end-date']),
  894. }
  895. )
  896. console.log('\nComparison Summary:')
  897. console.log('==================')
  898. console.log(`Total projects processed: ${totalProjects}`)
  899. console.log(`Projects with errors: ${totalErrors}`)
  900. if (totalErrors > 0) {
  901. throw new Error('Some project comparisons failed')
  902. }
  903. }
  904. async function main() {
  905. const options = handleOptions()
  906. await ensureGlobalBlobsLoaded()
  907. const projectId = options.projectId
  908. if (options.status) {
  909. await displayBackupStatus(projectId)
  910. } else if (options.list) {
  911. await displayPendingBackups(options)
  912. } else if (options.pending) {
  913. await backupPendingProjects(options)
  914. } else if (options.init) {
  915. await initializeProjects(options)
  916. } else if (options.compare) {
  917. if (options['start-date'] && options['end-date']) {
  918. await compareAllProjects(options)
  919. } else {
  920. await compareBackups(projectId, options)
  921. }
  922. } else {
  923. await backupProject(projectId, options)
  924. }
  925. }
  926. // Only run command-line interface when script is run directly
  927. if (import.meta.url === `file://${process.argv[1]}`) {
  928. main()
  929. .then(() => {
  930. console.log(
  931. gracefulShutdownInitiated ? 'Exited - graceful shutdown' : 'Completed'
  932. )
  933. })
  934. .catch(err => {
  935. console.error('Error backing up project:', err)
  936. process.exit(1)
  937. })
  938. .finally(() => {
  939. knex
  940. .destroy()
  941. .then(() => {
  942. console.log('Postgres connection closed')
  943. })
  944. .catch(err => {
  945. console.error('Error closing Postgres connection:', err)
  946. })
  947. client
  948. .close()
  949. .then(() => {
  950. console.log('MongoDB connection closed')
  951. })
  952. .catch(err => {
  953. console.error('Error closing MongoDB connection:', err)
  954. })
  955. })
  956. }