DocArchiveManager.js 6.7 KB

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