PackManager.js 32 KB

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