PackManager.js 32 KB

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