PackManager.js 32 KB

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