PackManager.js 34 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202
  1. /* eslint-disable
  2. no-unused-vars,
  3. */
  4. // TODO: This file was created by bulk-decaffeinate.
  5. // Fix any style issues and re-enable lint.
  6. /*
  7. * decaffeinate suggestions:
  8. * DS101: Remove unnecessary use of Array.from
  9. * DS102: Remove unnecessary code created because of implicit returns
  10. * DS205: Consider reworking code to avoid use of IIFEs
  11. * DS207: Consider shorter variations of null checks
  12. * Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md
  13. */
  14. let PackManager
  15. const async = require('async')
  16. const _ = require('underscore')
  17. const Bson = require('bson')
  18. const BSON = new Bson()
  19. const { db, ObjectId } = require('./mongodb')
  20. const logger = require('@overleaf/logger')
  21. const LockManager = require('./LockManager')
  22. const MongoAWS = require('./MongoAWS')
  23. const Metrics = require('@overleaf/metrics')
  24. const ProjectIterator = require('./ProjectIterator')
  25. const DocIterator = require('./DocIterator')
  26. const Settings = require('@overleaf/settings')
  27. const util = require('util')
  28. const keys = Settings.redis.lock.key_schema
  29. // Sharejs operations are stored in a 'pack' object
  30. //
  31. // e.g. a single sharejs update looks like
  32. //
  33. // {
  34. // "doc_id" : 549dae9e0a2a615c0c7f0c98,
  35. // "project_id" : 549dae9c0a2a615c0c7f0c8c,
  36. // "op" : [ {"p" : 6981, "d" : "?" } ],
  37. // "meta" : { "user_id" : 52933..., "start_ts" : 1422310693931, "end_ts" : 1422310693931 },
  38. // "v" : 17082
  39. // }
  40. //
  41. // and a pack looks like this
  42. //
  43. // {
  44. // "doc_id" : 549dae9e0a2a615c0c7f0c98,
  45. // "project_id" : 549dae9c0a2a615c0c7f0c8c,
  46. // "pack" : [ U1, U2, U3, ...., UN],
  47. // "meta" : { "user_id" : 52933..., "start_ts" : 1422310693931, "end_ts" : 1422310693931 },
  48. // "v" : 17082
  49. // "v_end" : ...
  50. // }
  51. //
  52. // where U1, U2, U3, .... are single updates stripped of their
  53. // doc_id and project_id fields (which are the same for all the
  54. // updates in the pack).
  55. //
  56. // The pack itself has v and meta fields, this makes it possible to
  57. // treat packs and single updates in a similar way.
  58. //
  59. // The v field of the pack itself is from the first entry U1, the
  60. // v_end field from UN. The meta.end_ts field of the pack itself is
  61. // from the last entry UN, the meta.start_ts field from U1.
  62. const DAYS = 24 * 3600 * 1000 // one day in milliseconds
  63. module.exports = PackManager = {
  64. MAX_SIZE: 1024 * 1024, // make these configurable parameters
  65. MAX_COUNT: 1024,
  66. insertCompressedUpdates(
  67. projectId,
  68. docId,
  69. lastUpdate,
  70. newUpdates,
  71. temporary,
  72. callback
  73. ) {
  74. if (callback == null) {
  75. callback = function () {}
  76. }
  77. if (newUpdates.length === 0) {
  78. return callback()
  79. }
  80. // never append permanent ops to a pack that will expire
  81. if (
  82. (lastUpdate != null ? lastUpdate.expiresAt : undefined) != null &&
  83. !temporary
  84. ) {
  85. lastUpdate = null
  86. }
  87. const updatesToFlush = []
  88. const updatesRemaining = newUpdates.slice()
  89. let n = (lastUpdate != null ? lastUpdate.n : undefined) || 0
  90. let sz = (lastUpdate != null ? lastUpdate.sz : undefined) || 0
  91. while (
  92. updatesRemaining.length &&
  93. n < PackManager.MAX_COUNT &&
  94. sz < PackManager.MAX_SIZE
  95. ) {
  96. const nextUpdate = updatesRemaining[0]
  97. const nextUpdateSize = BSON.calculateObjectSize(nextUpdate)
  98. if (nextUpdateSize + sz > PackManager.MAX_SIZE && n > 0) {
  99. break
  100. }
  101. n++
  102. sz += nextUpdateSize
  103. updatesToFlush.push(updatesRemaining.shift())
  104. }
  105. return PackManager.flushCompressedUpdates(
  106. projectId,
  107. docId,
  108. lastUpdate,
  109. updatesToFlush,
  110. temporary,
  111. function (error) {
  112. if (error != null) {
  113. return callback(error)
  114. }
  115. return PackManager.insertCompressedUpdates(
  116. projectId,
  117. docId,
  118. null,
  119. updatesRemaining,
  120. temporary,
  121. callback
  122. )
  123. }
  124. )
  125. },
  126. flushCompressedUpdates(
  127. projectId,
  128. docId,
  129. lastUpdate,
  130. newUpdates,
  131. temporary,
  132. callback
  133. ) {
  134. if (callback == null) {
  135. callback = function () {}
  136. }
  137. if (newUpdates.length === 0) {
  138. return callback()
  139. }
  140. let canAppend = false
  141. // check if it is safe to append to an existing pack
  142. if (lastUpdate != null) {
  143. if (!temporary && lastUpdate.expiresAt == null) {
  144. // permanent pack appends to permanent pack
  145. canAppend = true
  146. }
  147. const age =
  148. Date.now() -
  149. (lastUpdate.meta != null ? lastUpdate.meta.start_ts : undefined)
  150. if (temporary && lastUpdate.expiresAt != null && age < 1 * DAYS) {
  151. // temporary pack appends to temporary pack if same day
  152. canAppend = true
  153. }
  154. }
  155. if (canAppend) {
  156. return PackManager.appendUpdatesToExistingPack(
  157. projectId,
  158. docId,
  159. lastUpdate,
  160. newUpdates,
  161. temporary,
  162. callback
  163. )
  164. } else {
  165. return PackManager.insertUpdatesIntoNewPack(
  166. projectId,
  167. docId,
  168. newUpdates,
  169. temporary,
  170. callback
  171. )
  172. }
  173. },
  174. insertUpdatesIntoNewPack(projectId, docId, newUpdates, temporary, callback) {
  175. if (callback == null) {
  176. callback = function () {}
  177. }
  178. const first = newUpdates[0]
  179. const last = newUpdates[newUpdates.length - 1]
  180. const n = newUpdates.length
  181. const sz = BSON.calculateObjectSize(newUpdates)
  182. const newPack = {
  183. project_id: ObjectId(projectId.toString()),
  184. doc_id: ObjectId(docId.toString()),
  185. pack: newUpdates,
  186. n,
  187. sz,
  188. meta: {
  189. start_ts: first.meta.start_ts,
  190. end_ts: last.meta.end_ts,
  191. },
  192. v: first.v,
  193. v_end: last.v,
  194. temporary,
  195. }
  196. if (temporary) {
  197. newPack.expiresAt = new Date(Date.now() + 7 * DAYS)
  198. newPack.last_checked = new Date(Date.now() + 30 * DAYS) // never check temporary packs
  199. }
  200. logger.debug(
  201. { projectId, docId, newUpdates },
  202. 'inserting updates into new pack'
  203. )
  204. return db.docHistory.insertOne(newPack, function (err) {
  205. if (err != null) {
  206. return callback(err)
  207. }
  208. Metrics.inc(`insert-pack-${temporary ? 'temporary' : 'permanent'}`)
  209. if (temporary) {
  210. return callback()
  211. } else {
  212. return PackManager.updateIndex(projectId, docId, callback)
  213. }
  214. })
  215. },
  216. appendUpdatesToExistingPack(
  217. projectId,
  218. docId,
  219. lastUpdate,
  220. newUpdates,
  221. temporary,
  222. callback
  223. ) {
  224. if (callback == null) {
  225. callback = function () {}
  226. }
  227. const first = newUpdates[0]
  228. const last = newUpdates[newUpdates.length - 1]
  229. const n = newUpdates.length
  230. const sz = BSON.calculateObjectSize(newUpdates)
  231. const query = {
  232. _id: lastUpdate._id,
  233. project_id: ObjectId(projectId.toString()),
  234. doc_id: ObjectId(docId.toString()),
  235. pack: { $exists: true },
  236. }
  237. const update = {
  238. $push: {
  239. pack: { $each: newUpdates },
  240. },
  241. $inc: {
  242. n,
  243. sz,
  244. },
  245. $set: {
  246. 'meta.end_ts': last.meta.end_ts,
  247. v_end: last.v,
  248. },
  249. }
  250. if (lastUpdate.expiresAt && temporary) {
  251. update.$set.expiresAt = new Date(Date.now() + 7 * DAYS)
  252. }
  253. logger.debug(
  254. { projectId, docId, lastUpdate, newUpdates },
  255. 'appending updates to existing pack'
  256. )
  257. Metrics.inc(`append-pack-${temporary ? 'temporary' : 'permanent'}`)
  258. return db.docHistory.updateOne(query, update, callback)
  259. },
  260. // Retrieve all changes for a document
  261. getOpsByVersionRange(projectId, docId, fromVersion, toVersion, callback) {
  262. if (callback == null) {
  263. callback = function () {}
  264. }
  265. return PackManager.loadPacksByVersionRange(
  266. projectId,
  267. docId,
  268. fromVersion,
  269. toVersion,
  270. function (error) {
  271. if (error) return callback(error)
  272. const query = { doc_id: ObjectId(docId.toString()) }
  273. if (toVersion != null) {
  274. query.v = { $lte: toVersion }
  275. }
  276. if (fromVersion != null) {
  277. query.v_end = { $gte: fromVersion }
  278. }
  279. // console.log "query:", query
  280. return db.docHistory
  281. .find(query)
  282. .sort({ v: -1 })
  283. .toArray(function (err, result) {
  284. if (err != null) {
  285. return callback(err)
  286. }
  287. // console.log "getOpsByVersionRange:", err, result
  288. const updates = []
  289. const opInRange = function (op, from, to) {
  290. if (fromVersion != null && op.v < fromVersion) {
  291. return false
  292. }
  293. if (toVersion != null && op.v > toVersion) {
  294. return false
  295. }
  296. return true
  297. }
  298. for (const docHistory of Array.from(result)) {
  299. // console.log 'adding', docHistory.pack
  300. for (const op of Array.from(docHistory.pack.reverse())) {
  301. if (opInRange(op, fromVersion, toVersion)) {
  302. op.project_id = docHistory.project_id
  303. op.doc_id = docHistory.doc_id
  304. // console.log "added op", op.v, fromVersion, toVersion
  305. updates.push(op)
  306. }
  307. }
  308. }
  309. return callback(null, updates)
  310. })
  311. }
  312. )
  313. },
  314. loadPacksByVersionRange(projectId, docId, fromVersion, toVersion, callback) {
  315. return PackManager.getIndex(docId, function (err, indexResult) {
  316. let pack
  317. if (err != null) {
  318. return callback(err)
  319. }
  320. const indexPacks =
  321. (indexResult != null ? indexResult.packs : undefined) || []
  322. const packInRange = function (pack, from, to) {
  323. if (fromVersion != null && pack.v_end < fromVersion) {
  324. return false
  325. }
  326. if (toVersion != null && pack.v > toVersion) {
  327. return false
  328. }
  329. return true
  330. }
  331. const neededIds = (() => {
  332. const result = []
  333. for (pack of Array.from(indexPacks)) {
  334. if (packInRange(pack, fromVersion, toVersion)) {
  335. result.push(pack._id)
  336. }
  337. }
  338. return result
  339. })()
  340. if (neededIds.length) {
  341. return PackManager.fetchPacksIfNeeded(
  342. projectId,
  343. docId,
  344. neededIds,
  345. callback
  346. )
  347. } else {
  348. return callback()
  349. }
  350. })
  351. },
  352. fetchPacksIfNeeded(projectId, docId, packIds, callback) {
  353. let id
  354. return db.docHistory
  355. .find({ _id: { $in: packIds.map(ObjectId) } }, { projection: { _id: 1 } })
  356. .toArray(function (err, loadedPacks) {
  357. if (err != null) {
  358. return callback(err)
  359. }
  360. const allPackIds = (() => {
  361. const result1 = []
  362. for (id of Array.from(packIds)) {
  363. result1.push(id.toString())
  364. }
  365. return result1
  366. })()
  367. const loadedPackIds = Array.from(loadedPacks).map(pack =>
  368. pack._id.toString()
  369. )
  370. const packIdsToFetch = _.difference(allPackIds, loadedPackIds)
  371. logger.debug(
  372. { projectId, docId, loadedPackIds, allPackIds, packIdsToFetch },
  373. 'analysed packs'
  374. )
  375. if (packIdsToFetch.length === 0) {
  376. return callback()
  377. }
  378. return async.eachLimit(
  379. packIdsToFetch,
  380. 4,
  381. (packId, cb) => MongoAWS.unArchivePack(projectId, docId, packId, cb),
  382. function (err) {
  383. if (err != null) {
  384. return callback(err)
  385. }
  386. logger.debug({ projectId, docId }, 'done unarchiving')
  387. return callback()
  388. }
  389. )
  390. })
  391. },
  392. findAllDocsInProject(projectId, callback) {
  393. const docIdSet = new Set()
  394. async.series(
  395. [
  396. cb => {
  397. db.docHistory
  398. .find(
  399. { project_id: ObjectId(projectId) },
  400. { projection: { pack: false } }
  401. )
  402. .toArray((err, packs) => {
  403. if (err) return callback(err)
  404. packs.forEach(pack => {
  405. docIdSet.add(pack.doc_id.toString())
  406. })
  407. return cb()
  408. })
  409. },
  410. cb => {
  411. db.docHistoryIndex
  412. .find({ project_id: ObjectId(projectId) })
  413. .toArray((err, indexes) => {
  414. if (err) return callback(err)
  415. indexes.forEach(index => {
  416. docIdSet.add(index._id.toString())
  417. })
  418. return cb()
  419. })
  420. },
  421. ],
  422. err => {
  423. if (err) return callback(err)
  424. callback(null, [...docIdSet])
  425. }
  426. )
  427. },
  428. // rewrite any query using doc_id to use _id instead
  429. // (because docHistoryIndex uses the doc_id)
  430. _rewriteQueryForIndex(query) {
  431. const indexQuery = _.omit(query, 'doc_id')
  432. if ('doc_id' in query) {
  433. indexQuery._id = query.doc_id
  434. }
  435. return indexQuery
  436. },
  437. // Retrieve all changes across a project
  438. _findPacks(query, sortKeys, callback) {
  439. // get all the docHistory Entries
  440. return db.docHistory
  441. .find(query, { projection: { pack: false } })
  442. .sort(sortKeys)
  443. .toArray(function (err, packs) {
  444. let pack
  445. if (err != null) {
  446. return callback(err)
  447. }
  448. const allPacks = []
  449. const seenIds = {}
  450. for (pack of Array.from(packs)) {
  451. allPacks.push(pack)
  452. seenIds[pack._id] = true
  453. }
  454. const indexQuery = PackManager._rewriteQueryForIndex(query)
  455. return db.docHistoryIndex
  456. .find(indexQuery)
  457. .toArray(function (err, indexes) {
  458. if (err != null) {
  459. return callback(err)
  460. }
  461. for (const index of Array.from(indexes)) {
  462. for (pack of Array.from(index.packs)) {
  463. if (!seenIds[pack._id]) {
  464. pack.project_id = index.project_id
  465. pack.doc_id = index._id
  466. pack.fromIndex = true
  467. allPacks.push(pack)
  468. seenIds[pack._id] = true
  469. }
  470. }
  471. }
  472. return callback(null, allPacks)
  473. })
  474. })
  475. },
  476. makeProjectIterator(projectId, before, callback) {
  477. PackManager._findPacks(
  478. { project_id: ObjectId(projectId) },
  479. { 'meta.end_ts': -1 },
  480. function (err, allPacks) {
  481. if (err) return callback(err)
  482. callback(
  483. null,
  484. new ProjectIterator(allPacks, before, PackManager.getPackById)
  485. )
  486. }
  487. )
  488. },
  489. makeDocIterator(docId, callback) {
  490. PackManager._findPacks(
  491. { doc_id: ObjectId(docId) },
  492. { v: -1 },
  493. function (err, allPacks) {
  494. if (err) return callback(err)
  495. callback(null, new DocIterator(allPacks, PackManager.getPackById))
  496. }
  497. )
  498. },
  499. getPackById(projectId, docId, packId, callback) {
  500. return db.docHistory.findOne({ _id: packId }, function (err, pack) {
  501. if (err != null) {
  502. return callback(err)
  503. }
  504. if (pack == null) {
  505. return MongoAWS.unArchivePack(projectId, docId, packId, callback)
  506. } else if (pack.expiresAt != null && pack.temporary === false) {
  507. // we only need to touch the TTL when listing the changes in the project
  508. // because diffs on individual documents are always done after that
  509. return PackManager.increaseTTL(pack, callback)
  510. // only do this for cached packs, not temporary ones to avoid older packs
  511. // being kept longer than newer ones (which messes up the last update version)
  512. } else {
  513. return callback(null, pack)
  514. }
  515. })
  516. },
  517. increaseTTL(pack, callback) {
  518. if (pack.expiresAt < new Date(Date.now() + 6 * DAYS)) {
  519. // update cache expiry since we are using this pack
  520. return db.docHistory.updateOne(
  521. { _id: pack._id },
  522. { $set: { expiresAt: new Date(Date.now() + 7 * DAYS) } },
  523. err => callback(err, pack)
  524. )
  525. } else {
  526. return callback(null, pack)
  527. }
  528. },
  529. // Manage docHistoryIndex collection
  530. getIndex(docId, callback) {
  531. return db.docHistoryIndex.findOne(
  532. { _id: ObjectId(docId.toString()) },
  533. callback
  534. )
  535. },
  536. getPackFromIndex(docId, packId, callback) {
  537. return db.docHistoryIndex.findOne(
  538. { _id: ObjectId(docId.toString()), 'packs._id': packId },
  539. { projection: { 'packs.$': 1 } },
  540. callback
  541. )
  542. },
  543. getLastPackFromIndex(docId, callback) {
  544. return db.docHistoryIndex.findOne(
  545. { _id: ObjectId(docId.toString()) },
  546. { projection: { packs: { $slice: -1 } } },
  547. function (err, indexPack) {
  548. if (err != null) {
  549. return callback(err)
  550. }
  551. if (indexPack == null) {
  552. return callback()
  553. }
  554. return callback(null, indexPack[0])
  555. }
  556. )
  557. },
  558. getIndexWithKeys(docId, callback) {
  559. return PackManager.getIndex(docId, function (err, index) {
  560. if (err != null) {
  561. return callback(err)
  562. }
  563. if (index == null) {
  564. return callback()
  565. }
  566. for (const pack of Array.from(
  567. (index != null ? index.packs : undefined) || []
  568. )) {
  569. index[pack._id] = pack
  570. }
  571. return callback(null, index)
  572. })
  573. },
  574. initialiseIndex(projectId, docId, callback) {
  575. return PackManager.findCompletedPacks(
  576. projectId,
  577. docId,
  578. function (err, packs) {
  579. // console.log 'err', err, 'packs', packs, packs?.length
  580. if (err != null) {
  581. return callback(err)
  582. }
  583. if (packs == null) {
  584. return callback()
  585. }
  586. return PackManager.insertPacksIntoIndexWithLock(
  587. projectId,
  588. docId,
  589. packs,
  590. callback
  591. )
  592. }
  593. )
  594. },
  595. updateIndex(projectId, docId, callback) {
  596. // find all packs prior to current pack
  597. return PackManager.findUnindexedPacks(
  598. projectId,
  599. docId,
  600. function (err, newPacks) {
  601. if (err != null) {
  602. return callback(err)
  603. }
  604. if (newPacks == null || newPacks.length === 0) {
  605. return callback()
  606. }
  607. return PackManager.insertPacksIntoIndexWithLock(
  608. projectId,
  609. docId,
  610. newPacks,
  611. function (err) {
  612. if (err != null) {
  613. return callback(err)
  614. }
  615. logger.debug(
  616. { projectId, docId, newPacks },
  617. 'added new packs to index'
  618. )
  619. return callback()
  620. }
  621. )
  622. }
  623. )
  624. },
  625. findCompletedPacks(projectId, docId, callback) {
  626. const query = {
  627. doc_id: ObjectId(docId.toString()),
  628. expiresAt: { $exists: false },
  629. }
  630. return db.docHistory
  631. .find(query, { projection: { pack: false } })
  632. .sort({ v: 1 })
  633. .toArray(function (err, packs) {
  634. if (err != null) {
  635. return callback(err)
  636. }
  637. if (packs == null) {
  638. return callback()
  639. }
  640. if (!(packs != null ? packs.length : undefined)) {
  641. return callback()
  642. }
  643. const last = packs.pop() // discard the last pack, if it's still in progress
  644. if (last.finalised) {
  645. packs.push(last)
  646. } // it's finalised so we push it back to archive it
  647. return callback(null, packs)
  648. })
  649. },
  650. findPacks(projectId, docId, callback) {
  651. const query = {
  652. doc_id: ObjectId(docId.toString()),
  653. expiresAt: { $exists: false },
  654. }
  655. return db.docHistory
  656. .find(query, { projection: { pack: false } })
  657. .sort({ v: 1 })
  658. .toArray(function (err, packs) {
  659. if (err != null) {
  660. return callback(err)
  661. }
  662. if (packs == null) {
  663. return callback()
  664. }
  665. if (!(packs != null ? packs.length : undefined)) {
  666. return callback()
  667. }
  668. return callback(null, packs)
  669. })
  670. },
  671. findUnindexedPacks(projectId, docId, callback) {
  672. return PackManager.getIndexWithKeys(docId, function (err, indexResult) {
  673. if (err != null) {
  674. return callback(err)
  675. }
  676. return PackManager.findCompletedPacks(
  677. projectId,
  678. docId,
  679. function (err, historyPacks) {
  680. let pack
  681. if (err != null) {
  682. return callback(err)
  683. }
  684. if (historyPacks == null) {
  685. return callback()
  686. }
  687. // select only the new packs not already in the index
  688. let newPacks = (() => {
  689. const result = []
  690. for (pack of Array.from(historyPacks)) {
  691. if (
  692. (indexResult != null ? indexResult[pack._id] : undefined) ==
  693. null
  694. ) {
  695. result.push(pack)
  696. }
  697. }
  698. return result
  699. })()
  700. newPacks = (() => {
  701. const result1 = []
  702. for (pack of Array.from(newPacks)) {
  703. result1.push(
  704. _.omit(
  705. pack,
  706. 'doc_id',
  707. 'project_id',
  708. 'n',
  709. 'sz',
  710. 'last_checked',
  711. 'finalised'
  712. )
  713. )
  714. }
  715. return result1
  716. })()
  717. if (newPacks.length) {
  718. logger.debug(
  719. { projectId, docId, n: newPacks.length },
  720. 'found new packs'
  721. )
  722. }
  723. return callback(null, newPacks)
  724. }
  725. )
  726. })
  727. },
  728. insertPacksIntoIndexWithLock(projectId, docId, newPacks, callback) {
  729. return LockManager.runWithLock(
  730. keys.historyIndexLock({ doc_id: docId }),
  731. releaseLock =>
  732. PackManager._insertPacksIntoIndex(
  733. projectId,
  734. docId,
  735. newPacks,
  736. releaseLock
  737. ),
  738. callback
  739. )
  740. },
  741. _insertPacksIntoIndex(projectId, docId, newPacks, callback) {
  742. return db.docHistoryIndex.updateOne(
  743. { _id: ObjectId(docId.toString()) },
  744. {
  745. $setOnInsert: { project_id: ObjectId(projectId.toString()) },
  746. $push: {
  747. packs: { $each: newPacks, $sort: { v: 1 } },
  748. },
  749. },
  750. {
  751. upsert: true,
  752. },
  753. callback
  754. )
  755. },
  756. // Archiving packs to S3
  757. archivePack(projectId, docId, packId, callback) {
  758. const clearFlagOnError = function (err, cb) {
  759. if (err != null) {
  760. // clear the inS3 flag on error
  761. return PackManager.clearPackAsArchiveInProgress(
  762. projectId,
  763. docId,
  764. packId,
  765. function (err2) {
  766. if (err2 != null) {
  767. return cb(err2)
  768. }
  769. return cb(err)
  770. }
  771. )
  772. } else {
  773. return cb()
  774. }
  775. }
  776. return async.series(
  777. [
  778. cb =>
  779. PackManager.checkArchiveNotInProgress(projectId, docId, packId, cb),
  780. cb =>
  781. PackManager.markPackAsArchiveInProgress(projectId, docId, packId, cb),
  782. cb =>
  783. MongoAWS.archivePack(projectId, docId, packId, err =>
  784. clearFlagOnError(err, cb)
  785. ),
  786. cb =>
  787. PackManager.checkArchivedPack(projectId, docId, packId, err =>
  788. clearFlagOnError(err, cb)
  789. ),
  790. cb => PackManager.markPackAsArchived(projectId, docId, packId, cb),
  791. cb =>
  792. PackManager.setTTLOnArchivedPack(projectId, docId, packId, callback),
  793. ],
  794. callback
  795. )
  796. },
  797. checkArchivedPack(projectId, docId, packId, callback) {
  798. return db.docHistory.findOne({ _id: packId }, function (err, pack) {
  799. if (err != null) {
  800. return callback(err)
  801. }
  802. if (pack == null) {
  803. return callback(new Error('pack not found'))
  804. }
  805. return MongoAWS.readArchivedPack(
  806. projectId,
  807. docId,
  808. packId,
  809. function (err, result) {
  810. if (err) return callback(err)
  811. delete result.last_checked
  812. delete pack.last_checked
  813. // need to compare ids as ObjectIds with .equals()
  814. for (const key of ['_id', 'project_id', 'doc_id']) {
  815. if (result[key].equals(pack[key])) {
  816. result[key] = pack[key]
  817. }
  818. }
  819. for (let i = 0; i < result.pack.length; i++) {
  820. const op = result.pack[i]
  821. if (op._id != null && op._id.equals(pack.pack[i]._id)) {
  822. op._id = pack.pack[i]._id
  823. }
  824. }
  825. if (_.isEqual(pack, result)) {
  826. return callback()
  827. } else {
  828. logger.err(
  829. {
  830. pack,
  831. result,
  832. jsondiff: JSON.stringify(pack) === JSON.stringify(result),
  833. },
  834. 'difference when comparing packs'
  835. )
  836. return callback(
  837. new Error('pack retrieved from s3 does not match pack in mongo')
  838. )
  839. }
  840. }
  841. )
  842. })
  843. },
  844. // Extra methods to test archive/unarchive for a doc_id
  845. pushOldPacks(projectId, docId, callback) {
  846. return PackManager.findPacks(projectId, docId, function (err, packs) {
  847. if (err != null) {
  848. return callback(err)
  849. }
  850. if (!(packs != null ? packs.length : undefined)) {
  851. return callback()
  852. }
  853. return PackManager.processOldPack(
  854. projectId,
  855. docId,
  856. packs[0]._id,
  857. callback
  858. )
  859. })
  860. },
  861. pullOldPacks(projectId, docId, callback) {
  862. return PackManager.loadPacksByVersionRange(
  863. projectId,
  864. docId,
  865. null,
  866. null,
  867. callback
  868. )
  869. },
  870. // Processing old packs via worker
  871. processOldPack(projectId, docId, packId, callback) {
  872. const markAsChecked = err =>
  873. PackManager.markPackAsChecked(projectId, docId, packId, function (err2) {
  874. if (err2 != null) {
  875. return callback(err2)
  876. }
  877. return callback(err)
  878. })
  879. logger.debug({ projectId, docId }, 'processing old packs')
  880. return db.docHistory.findOne({ _id: packId }, function (err, pack) {
  881. if (err != null) {
  882. return markAsChecked(err)
  883. }
  884. if (pack == null) {
  885. return markAsChecked()
  886. }
  887. if (pack.expiresAt != null) {
  888. return callback()
  889. } // return directly
  890. return PackManager.finaliseIfNeeded(
  891. projectId,
  892. docId,
  893. pack._id,
  894. pack,
  895. function (err) {
  896. if (err != null) {
  897. return markAsChecked(err)
  898. }
  899. return PackManager.updateIndexIfNeeded(
  900. projectId,
  901. docId,
  902. function (err) {
  903. if (err != null) {
  904. return markAsChecked(err)
  905. }
  906. return PackManager.findUnarchivedPacks(
  907. projectId,
  908. docId,
  909. function (err, unarchivedPacks) {
  910. if (err != null) {
  911. return markAsChecked(err)
  912. }
  913. if (
  914. !(unarchivedPacks != null
  915. ? unarchivedPacks.length
  916. : undefined)
  917. ) {
  918. logger.debug(
  919. { projectId, docId },
  920. 'no packs need archiving'
  921. )
  922. return markAsChecked()
  923. }
  924. return async.eachSeries(
  925. unarchivedPacks,
  926. (pack, cb) =>
  927. PackManager.archivePack(projectId, docId, pack._id, cb),
  928. function (err) {
  929. if (err != null) {
  930. return markAsChecked(err)
  931. }
  932. logger.debug({ projectId, docId }, 'done processing')
  933. return markAsChecked()
  934. }
  935. )
  936. }
  937. )
  938. }
  939. )
  940. }
  941. )
  942. })
  943. },
  944. finaliseIfNeeded(projectId, docId, packId, pack, callback) {
  945. const sz = pack.sz / (1024 * 1024) // in fractions of a megabyte
  946. const n = pack.n / 1024 // in fraction of 1024 ops
  947. const age = (Date.now() - pack.meta.end_ts) / DAYS
  948. if (age < 30) {
  949. // always keep if less than 1 month old
  950. logger.debug({ projectId, docId, packId, age }, 'less than 30 days old')
  951. return callback()
  952. }
  953. // compute an archiving threshold which decreases for each month of age
  954. const archiveThreshold = 30 / age
  955. if (sz > archiveThreshold || n > archiveThreshold || age > 90) {
  956. logger.debug(
  957. { projectId, docId, packId, age, archiveThreshold, sz, n },
  958. 'meets archive threshold'
  959. )
  960. return PackManager.markPackAsFinalisedWithLock(
  961. projectId,
  962. docId,
  963. packId,
  964. callback
  965. )
  966. } else {
  967. logger.debug(
  968. { projectId, docId, packId, age, archiveThreshold, sz, n },
  969. 'does not meet archive threshold'
  970. )
  971. return callback()
  972. }
  973. },
  974. markPackAsFinalisedWithLock(projectId, docId, packId, callback) {
  975. return LockManager.runWithLock(
  976. keys.historyLock({ doc_id: docId }),
  977. releaseLock =>
  978. PackManager._markPackAsFinalised(projectId, docId, packId, releaseLock),
  979. callback
  980. )
  981. },
  982. _markPackAsFinalised(projectId, docId, packId, callback) {
  983. logger.debug({ projectId, docId, packId }, 'marking pack as finalised')
  984. return db.docHistory.updateOne(
  985. { _id: packId },
  986. { $set: { finalised: true } },
  987. callback
  988. )
  989. },
  990. updateIndexIfNeeded(projectId, docId, callback) {
  991. logger.debug({ projectId, docId }, 'archiving old packs')
  992. return PackManager.getIndexWithKeys(docId, function (err, index) {
  993. if (err != null) {
  994. return callback(err)
  995. }
  996. if (index == null) {
  997. return PackManager.initialiseIndex(projectId, docId, callback)
  998. } else {
  999. return PackManager.updateIndex(projectId, docId, callback)
  1000. }
  1001. })
  1002. },
  1003. markPackAsChecked(projectId, docId, packId, callback) {
  1004. logger.debug({ projectId, docId, packId }, 'marking pack as checked')
  1005. return db.docHistory.updateOne(
  1006. { _id: packId },
  1007. { $currentDate: { last_checked: true } },
  1008. callback
  1009. )
  1010. },
  1011. findUnarchivedPacks(projectId, docId, callback) {
  1012. return PackManager.getIndex(docId, function (err, indexResult) {
  1013. if (err != null) {
  1014. return callback(err)
  1015. }
  1016. const indexPacks =
  1017. (indexResult != null ? indexResult.packs : undefined) || []
  1018. const unArchivedPacks = (() => {
  1019. const result = []
  1020. for (const pack of Array.from(indexPacks)) {
  1021. if (pack.inS3 == null) {
  1022. result.push(pack)
  1023. }
  1024. }
  1025. return result
  1026. })()
  1027. if (unArchivedPacks.length) {
  1028. logger.debug(
  1029. { projectId, docId, n: unArchivedPacks.length },
  1030. 'find unarchived packs'
  1031. )
  1032. }
  1033. return callback(null, unArchivedPacks)
  1034. })
  1035. },
  1036. // Archive locking flags
  1037. checkArchiveNotInProgress(projectId, docId, packId, callback) {
  1038. logger.debug(
  1039. { projectId, docId, packId },
  1040. 'checking if archive in progress'
  1041. )
  1042. return PackManager.getPackFromIndex(docId, packId, function (err, result) {
  1043. if (err != null) {
  1044. return callback(err)
  1045. }
  1046. if (result == null) {
  1047. return callback(new Error('pack not found in index'))
  1048. }
  1049. if (result.inS3) {
  1050. return callback(new Error('pack archiving already done'))
  1051. } else if (result.inS3 != null) {
  1052. return callback(new Error('pack archiving already in progress'))
  1053. } else {
  1054. return callback()
  1055. }
  1056. })
  1057. },
  1058. markPackAsArchiveInProgress(projectId, docId, packId, callback) {
  1059. logger.debug(
  1060. { projectId, docId },
  1061. 'marking pack as archive in progress status'
  1062. )
  1063. return db.docHistoryIndex.findOneAndUpdate(
  1064. {
  1065. _id: ObjectId(docId.toString()),
  1066. packs: { $elemMatch: { _id: packId, inS3: { $exists: false } } },
  1067. },
  1068. { $set: { 'packs.$.inS3': false } },
  1069. { projection: { 'packs.$': 1 } },
  1070. function (err, result) {
  1071. if (err != null) {
  1072. return callback(err)
  1073. }
  1074. if (!result.value) {
  1075. return callback(new Error('archive is already in progress'))
  1076. }
  1077. logger.debug(
  1078. { projectId, docId, packId },
  1079. 'marked as archive in progress'
  1080. )
  1081. return callback()
  1082. }
  1083. )
  1084. },
  1085. clearPackAsArchiveInProgress(projectId, docId, packId, callback) {
  1086. logger.debug(
  1087. { projectId, docId, packId },
  1088. 'clearing as archive in progress'
  1089. )
  1090. return db.docHistoryIndex.updateOne(
  1091. {
  1092. _id: ObjectId(docId.toString()),
  1093. packs: { $elemMatch: { _id: packId, inS3: false } },
  1094. },
  1095. { $unset: { 'packs.$.inS3': true } },
  1096. callback
  1097. )
  1098. },
  1099. markPackAsArchived(projectId, docId, packId, callback) {
  1100. logger.debug({ projectId, docId, packId }, 'marking pack as archived')
  1101. return db.docHistoryIndex.findOneAndUpdate(
  1102. {
  1103. _id: ObjectId(docId.toString()),
  1104. packs: { $elemMatch: { _id: packId, inS3: false } },
  1105. },
  1106. { $set: { 'packs.$.inS3': true } },
  1107. { projection: { 'packs.$': 1 } },
  1108. function (err, result) {
  1109. if (err != null) {
  1110. return callback(err)
  1111. }
  1112. if (!result.value) {
  1113. return callback(new Error('archive is not marked as progress'))
  1114. }
  1115. logger.debug({ projectId, docId, packId }, 'marked as archived')
  1116. return callback()
  1117. }
  1118. )
  1119. },
  1120. setTTLOnArchivedPack(projectId, docId, packId, callback) {
  1121. return db.docHistory.updateOne(
  1122. { _id: packId },
  1123. { $set: { expiresAt: new Date(Date.now() + 1 * DAYS) } },
  1124. function (err) {
  1125. if (err) {
  1126. return callback(err)
  1127. }
  1128. logger.debug({ projectId, docId, packId }, 'set expiry on pack')
  1129. return callback()
  1130. }
  1131. )
  1132. },
  1133. }
  1134. module.exports.promises = {
  1135. getOpsByVersionRange: util.promisify(PackManager.getOpsByVersionRange),
  1136. findAllDocsInProject: util.promisify(PackManager.findAllDocsInProject),
  1137. makeDocIterator: util.promisify(PackManager.makeDocIterator),
  1138. }
  1139. // _getOneDayInFutureWithRandomDelay: ->
  1140. // thirtyMins = 1000 * 60 * 30
  1141. // randomThirtyMinMax = Math.ceil(Math.random() * thirtyMins)
  1142. // return new Date(Date.now() + randomThirtyMinMax + 1*DAYS)