PackManager.js 32 KB

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