DocArchiveManager.js 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222
  1. const { callbackify } = require('util')
  2. const MongoManager = require('./MongoManager').promises
  3. const Errors = require('./Errors')
  4. const logger = require('@overleaf/logger')
  5. const Settings = require('@overleaf/settings')
  6. const crypto = require('crypto')
  7. const Streamifier = require('streamifier')
  8. const RangeManager = require('./RangeManager')
  9. const PersistorManager = require('./PersistorManager')
  10. const pMap = require('p-map')
  11. const Bson = require('bson')
  12. const BSON = new Bson()
  13. const PARALLEL_JOBS = Settings.parallelArchiveJobs
  14. const UN_ARCHIVE_BATCH_SIZE = Settings.unArchiveBatchSize
  15. module.exports = {
  16. archiveAllDocs: callbackify(archiveAllDocs),
  17. archiveDoc: callbackify(archiveDoc),
  18. unArchiveAllDocs: callbackify(unArchiveAllDocs),
  19. unarchiveDoc: callbackify(unarchiveDoc),
  20. destroyProject: callbackify(destroyProject),
  21. getDoc: callbackify(getDoc),
  22. promises: {
  23. archiveAllDocs,
  24. archiveDoc,
  25. unArchiveAllDocs,
  26. unarchiveDoc,
  27. destroyProject,
  28. getDoc,
  29. },
  30. }
  31. async function archiveAllDocs(projectId) {
  32. const docIds = await MongoManager.getNonArchivedProjectDocIds(projectId)
  33. await pMap(docIds, docId => archiveDoc(projectId, docId), {
  34. concurrency: PARALLEL_JOBS,
  35. })
  36. }
  37. async function archiveDoc(projectId, docId) {
  38. const doc = await MongoManager.getDocForArchiving(projectId, docId)
  39. if (!doc) {
  40. // The doc wasn't found, it was already archived, or the lock couldn't be
  41. // acquired. Since we don't know which it is, silently return.
  42. return
  43. }
  44. logger.debug(
  45. { project_id: projectId, doc_id: doc._id },
  46. 'sending doc to persistor'
  47. )
  48. const key = `${projectId}/${doc._id}`
  49. if (doc.lines == null) {
  50. throw new Error('doc has no lines')
  51. }
  52. // warn about any oversized docs already in mongo
  53. const linesSize = BSON.calculateObjectSize(doc.lines || {})
  54. const rangesSize = BSON.calculateObjectSize(doc.ranges || {})
  55. if (
  56. linesSize > Settings.max_doc_length ||
  57. rangesSize > Settings.max_doc_length
  58. ) {
  59. logger.warn(
  60. { project_id: projectId, doc_id: doc._id, linesSize, rangesSize },
  61. 'large doc found when archiving project'
  62. )
  63. }
  64. const json = JSON.stringify({
  65. lines: doc.lines,
  66. ranges: doc.ranges,
  67. rev: doc.rev,
  68. schema_v: 1,
  69. })
  70. // this should never happen, but protects against memory-corruption errors that
  71. // have happened in the past
  72. if (json.indexOf('\u0000') > -1) {
  73. const error = new Error('null bytes detected')
  74. logger.err({ err: error, doc }, error.message)
  75. throw error
  76. }
  77. const md5 = crypto.createHash('md5').update(json).digest('hex')
  78. const stream = Streamifier.createReadStream(json)
  79. await PersistorManager.sendStream(Settings.docstore.bucket, key, stream, {
  80. sourceMd5: md5,
  81. })
  82. await MongoManager.markDocAsArchived(projectId, docId, doc.rev)
  83. }
  84. async function unArchiveAllDocs(projectId) {
  85. while (true) {
  86. let docs
  87. if (Settings.docstore.keepSoftDeletedDocsArchived) {
  88. docs = await MongoManager.getNonDeletedArchivedProjectDocs(
  89. projectId,
  90. UN_ARCHIVE_BATCH_SIZE
  91. )
  92. } else {
  93. docs = await MongoManager.getArchivedProjectDocs(
  94. projectId,
  95. UN_ARCHIVE_BATCH_SIZE
  96. )
  97. }
  98. if (!docs || docs.length === 0) {
  99. break
  100. }
  101. await pMap(docs, doc => unarchiveDoc(projectId, doc._id), {
  102. concurrency: PARALLEL_JOBS,
  103. })
  104. }
  105. }
  106. // get the doc from the PersistorManager without storing it in mongo
  107. async function getDoc(projectId, docId) {
  108. const key = `${projectId}/${docId}`
  109. const sourceMd5 = await PersistorManager.getObjectMd5Hash(
  110. Settings.docstore.bucket,
  111. key
  112. )
  113. const stream = await PersistorManager.getObjectStream(
  114. Settings.docstore.bucket,
  115. key
  116. )
  117. stream.resume()
  118. const buffer = await _streamToBuffer(stream)
  119. const md5 = crypto.createHash('md5').update(buffer).digest('hex')
  120. if (sourceMd5 !== md5) {
  121. throw new Errors.Md5MismatchError('md5 mismatch when downloading doc', {
  122. key,
  123. sourceMd5,
  124. md5,
  125. })
  126. }
  127. const json = buffer.toString()
  128. return _deserializeArchivedDoc(json)
  129. }
  130. // get the doc and unarchive it to mongo
  131. async function unarchiveDoc(projectId, docId) {
  132. logger.debug({ projectId, docId }, 'getting doc from persistor')
  133. const mongoDoc = await MongoManager.findDoc(projectId, docId, {
  134. inS3: 1,
  135. rev: 1,
  136. })
  137. if (!mongoDoc.inS3) {
  138. // The doc is already unarchived
  139. return
  140. }
  141. const archivedDoc = await getDoc(projectId, docId)
  142. if (archivedDoc.rev == null) {
  143. // Older archived docs didn't have a rev. Assume that the rev of the
  144. // archived doc is the rev that was stored in Mongo when we retrieved it
  145. // earlier.
  146. archivedDoc.rev = mongoDoc.rev
  147. }
  148. await MongoManager.restoreArchivedDoc(projectId, docId, archivedDoc)
  149. }
  150. async function destroyProject(projectId) {
  151. const tasks = [MongoManager.destroyProject(projectId)]
  152. if (_isArchivingEnabled()) {
  153. tasks.push(
  154. PersistorManager.deleteDirectory(Settings.docstore.bucket, projectId)
  155. )
  156. }
  157. await Promise.all(tasks)
  158. }
  159. async function _streamToBuffer(stream) {
  160. const chunks = []
  161. return new Promise((resolve, reject) => {
  162. stream.on('data', chunk => chunks.push(chunk))
  163. stream.on('error', reject)
  164. stream.on('end', () => resolve(Buffer.concat(chunks)))
  165. })
  166. }
  167. function _deserializeArchivedDoc(json) {
  168. const doc = JSON.parse(json)
  169. const result = {}
  170. if (doc.schema_v === 1 && doc.lines != null) {
  171. result.lines = doc.lines
  172. if (doc.ranges != null) {
  173. result.ranges = RangeManager.jsonRangesToMongo(doc.ranges)
  174. }
  175. } else if (Array.isArray(doc)) {
  176. result.lines = doc
  177. } else {
  178. throw new Error("I don't understand the doc format in s3")
  179. }
  180. if (doc.rev != null) {
  181. result.rev = doc.rev
  182. }
  183. return result
  184. }
  185. function _isArchivingEnabled() {
  186. const backend = Settings.docstore.backend
  187. if (!backend) {
  188. return false
  189. }
  190. // The default backend is S3. If another backend is configured or the S3
  191. // backend itself is correctly configured, then archiving is enabled.
  192. if (backend === 's3' && Settings.docstore.s3 == null) {
  193. return false
  194. }
  195. return true
  196. }