PackManager.js 32 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171
  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('@overleaf/settings')
  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(
  528. project_id,
  529. doc_id,
  530. function (err, packs) {
  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. },
  547. updateIndex(project_id, doc_id, callback) {
  548. // find all packs prior to current pack
  549. return PackManager.findUnindexedPacks(
  550. project_id,
  551. doc_id,
  552. function (err, newPacks) {
  553. if (err != null) {
  554. return callback(err)
  555. }
  556. if (newPacks == null || newPacks.length === 0) {
  557. return callback()
  558. }
  559. return PackManager.insertPacksIntoIndexWithLock(
  560. project_id,
  561. doc_id,
  562. newPacks,
  563. function (err) {
  564. if (err != null) {
  565. return callback(err)
  566. }
  567. logger.log(
  568. { project_id, doc_id, newPacks },
  569. 'added new packs to index'
  570. )
  571. return callback()
  572. }
  573. )
  574. }
  575. )
  576. },
  577. findCompletedPacks(project_id, doc_id, callback) {
  578. const query = {
  579. doc_id: ObjectId(doc_id.toString()),
  580. expiresAt: { $exists: false },
  581. }
  582. return db.docHistory
  583. .find(query, { projection: { pack: false } })
  584. .sort({ v: 1 })
  585. .toArray(function (err, packs) {
  586. if (err != null) {
  587. return callback(err)
  588. }
  589. if (packs == null) {
  590. return callback()
  591. }
  592. if (!(packs != null ? packs.length : undefined)) {
  593. return callback()
  594. }
  595. const last = packs.pop() // discard the last pack, if it's still in progress
  596. if (last.finalised) {
  597. packs.push(last)
  598. } // it's finalised so we push it back to archive it
  599. return callback(null, packs)
  600. })
  601. },
  602. findPacks(project_id, doc_id, callback) {
  603. const query = {
  604. doc_id: ObjectId(doc_id.toString()),
  605. expiresAt: { $exists: false },
  606. }
  607. return db.docHistory
  608. .find(query, { projection: { pack: false } })
  609. .sort({ v: 1 })
  610. .toArray(function (err, packs) {
  611. if (err != null) {
  612. return callback(err)
  613. }
  614. if (packs == null) {
  615. return callback()
  616. }
  617. if (!(packs != null ? packs.length : undefined)) {
  618. return callback()
  619. }
  620. return callback(null, packs)
  621. })
  622. },
  623. findUnindexedPacks(project_id, doc_id, callback) {
  624. return PackManager.getIndexWithKeys(doc_id, function (err, indexResult) {
  625. if (err != null) {
  626. return callback(err)
  627. }
  628. return PackManager.findCompletedPacks(
  629. project_id,
  630. doc_id,
  631. function (err, historyPacks) {
  632. let pack
  633. if (err != null) {
  634. return callback(err)
  635. }
  636. if (historyPacks == null) {
  637. return callback()
  638. }
  639. // select only the new packs not already in the index
  640. let newPacks = (() => {
  641. const result = []
  642. for (pack of Array.from(historyPacks)) {
  643. if (
  644. (indexResult != null ? indexResult[pack._id] : undefined) ==
  645. null
  646. ) {
  647. result.push(pack)
  648. }
  649. }
  650. return result
  651. })()
  652. newPacks = (() => {
  653. const result1 = []
  654. for (pack of Array.from(newPacks)) {
  655. result1.push(
  656. _.omit(
  657. pack,
  658. 'doc_id',
  659. 'project_id',
  660. 'n',
  661. 'sz',
  662. 'last_checked',
  663. 'finalised'
  664. )
  665. )
  666. }
  667. return result1
  668. })()
  669. if (newPacks.length) {
  670. logger.log(
  671. { project_id, doc_id, n: newPacks.length },
  672. 'found new packs'
  673. )
  674. }
  675. return callback(null, newPacks)
  676. }
  677. )
  678. })
  679. },
  680. insertPacksIntoIndexWithLock(project_id, doc_id, newPacks, callback) {
  681. return LockManager.runWithLock(
  682. keys.historyIndexLock({ doc_id }),
  683. releaseLock =>
  684. PackManager._insertPacksIntoIndex(
  685. project_id,
  686. doc_id,
  687. newPacks,
  688. releaseLock
  689. ),
  690. callback
  691. )
  692. },
  693. _insertPacksIntoIndex(project_id, doc_id, newPacks, callback) {
  694. return db.docHistoryIndex.updateOne(
  695. { _id: ObjectId(doc_id.toString()) },
  696. {
  697. $setOnInsert: { project_id: ObjectId(project_id.toString()) },
  698. $push: {
  699. packs: { $each: newPacks, $sort: { v: 1 } },
  700. },
  701. },
  702. {
  703. upsert: true,
  704. },
  705. callback
  706. )
  707. },
  708. // Archiving packs to S3
  709. archivePack(project_id, doc_id, pack_id, callback) {
  710. const clearFlagOnError = function (err, cb) {
  711. if (err != null) {
  712. // clear the inS3 flag on error
  713. return PackManager.clearPackAsArchiveInProgress(
  714. project_id,
  715. doc_id,
  716. pack_id,
  717. function (err2) {
  718. if (err2 != null) {
  719. return cb(err2)
  720. }
  721. return cb(err)
  722. }
  723. )
  724. } else {
  725. return cb()
  726. }
  727. }
  728. return async.series(
  729. [
  730. cb =>
  731. PackManager.checkArchiveNotInProgress(
  732. project_id,
  733. doc_id,
  734. pack_id,
  735. cb
  736. ),
  737. cb =>
  738. PackManager.markPackAsArchiveInProgress(
  739. project_id,
  740. doc_id,
  741. pack_id,
  742. cb
  743. ),
  744. cb =>
  745. MongoAWS.archivePack(project_id, doc_id, pack_id, err =>
  746. clearFlagOnError(err, cb)
  747. ),
  748. cb =>
  749. PackManager.checkArchivedPack(project_id, doc_id, pack_id, err =>
  750. clearFlagOnError(err, cb)
  751. ),
  752. cb => PackManager.markPackAsArchived(project_id, doc_id, pack_id, cb),
  753. cb =>
  754. PackManager.setTTLOnArchivedPack(
  755. project_id,
  756. doc_id,
  757. pack_id,
  758. callback
  759. ),
  760. ],
  761. callback
  762. )
  763. },
  764. checkArchivedPack(project_id, doc_id, pack_id, callback) {
  765. return db.docHistory.findOne({ _id: pack_id }, function (err, pack) {
  766. if (err != null) {
  767. return callback(err)
  768. }
  769. if (pack == null) {
  770. return callback(new Error('pack not found'))
  771. }
  772. return MongoAWS.readArchivedPack(
  773. project_id,
  774. doc_id,
  775. pack_id,
  776. function (err, result) {
  777. delete result.last_checked
  778. delete pack.last_checked
  779. // need to compare ids as ObjectIds with .equals()
  780. for (const key of ['_id', 'project_id', 'doc_id']) {
  781. if (result[key].equals(pack[key])) {
  782. result[key] = pack[key]
  783. }
  784. }
  785. for (let i = 0; i < result.pack.length; i++) {
  786. const op = result.pack[i]
  787. if (op._id != null && op._id.equals(pack.pack[i]._id)) {
  788. op._id = pack.pack[i]._id
  789. }
  790. }
  791. if (_.isEqual(pack, result)) {
  792. return callback()
  793. } else {
  794. logger.err(
  795. {
  796. pack,
  797. result,
  798. jsondiff: JSON.stringify(pack) === JSON.stringify(result),
  799. },
  800. 'difference when comparing packs'
  801. )
  802. return callback(
  803. new Error('pack retrieved from s3 does not match pack in mongo')
  804. )
  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(
  840. project_id,
  841. doc_id,
  842. pack_id,
  843. function (err2) {
  844. if (err2 != null) {
  845. return callback(err2)
  846. }
  847. return callback(err)
  848. }
  849. )
  850. logger.log({ project_id, doc_id }, 'processing old packs')
  851. return db.docHistory.findOne({ _id: pack_id }, function (err, pack) {
  852. if (err != null) {
  853. return markAsChecked(err)
  854. }
  855. if (pack == null) {
  856. return markAsChecked()
  857. }
  858. if (pack.expiresAt != null) {
  859. return callback()
  860. } // return directly
  861. return PackManager.finaliseIfNeeded(
  862. project_id,
  863. doc_id,
  864. pack._id,
  865. pack,
  866. function (err) {
  867. if (err != null) {
  868. return markAsChecked(err)
  869. }
  870. return PackManager.updateIndexIfNeeded(
  871. project_id,
  872. doc_id,
  873. function (err) {
  874. if (err != null) {
  875. return markAsChecked(err)
  876. }
  877. return PackManager.findUnarchivedPacks(
  878. project_id,
  879. doc_id,
  880. function (err, unarchivedPacks) {
  881. if (err != null) {
  882. return markAsChecked(err)
  883. }
  884. if (
  885. !(unarchivedPacks != null
  886. ? unarchivedPacks.length
  887. : undefined)
  888. ) {
  889. logger.log(
  890. { project_id, doc_id },
  891. 'no packs need archiving'
  892. )
  893. return markAsChecked()
  894. }
  895. return async.eachSeries(
  896. unarchivedPacks,
  897. (pack, cb) =>
  898. PackManager.archivePack(project_id, doc_id, pack._id, cb),
  899. function (err) {
  900. if (err != null) {
  901. return markAsChecked(err)
  902. }
  903. logger.log({ project_id, doc_id }, 'done processing')
  904. return markAsChecked()
  905. }
  906. )
  907. }
  908. )
  909. }
  910. )
  911. }
  912. )
  913. })
  914. },
  915. finaliseIfNeeded(project_id, doc_id, pack_id, pack, callback) {
  916. const sz = pack.sz / (1024 * 1024) // in fractions of a megabyte
  917. const n = pack.n / 1024 // in fraction of 1024 ops
  918. const age = (Date.now() - pack.meta.end_ts) / DAYS
  919. if (age < 30) {
  920. // always keep if less than 1 month old
  921. logger.log({ project_id, doc_id, pack_id, age }, 'less than 30 days old')
  922. return callback()
  923. }
  924. // compute an archiving threshold which decreases for each month of age
  925. const archive_threshold = 30 / age
  926. if (sz > archive_threshold || n > archive_threshold || age > 90) {
  927. logger.log(
  928. { project_id, doc_id, pack_id, age, archive_threshold, sz, n },
  929. 'meets archive threshold'
  930. )
  931. return PackManager.markPackAsFinalisedWithLock(
  932. project_id,
  933. doc_id,
  934. pack_id,
  935. callback
  936. )
  937. } else {
  938. logger.log(
  939. { project_id, doc_id, pack_id, age, archive_threshold, sz, n },
  940. 'does not meet archive threshold'
  941. )
  942. return callback()
  943. }
  944. },
  945. markPackAsFinalisedWithLock(project_id, doc_id, pack_id, callback) {
  946. return LockManager.runWithLock(
  947. keys.historyLock({ doc_id }),
  948. releaseLock =>
  949. PackManager._markPackAsFinalised(
  950. project_id,
  951. doc_id,
  952. pack_id,
  953. releaseLock
  954. ),
  955. callback
  956. )
  957. },
  958. _markPackAsFinalised(project_id, doc_id, pack_id, callback) {
  959. logger.log({ project_id, doc_id, pack_id }, 'marking pack as finalised')
  960. return db.docHistory.updateOne(
  961. { _id: pack_id },
  962. { $set: { finalised: true } },
  963. callback
  964. )
  965. },
  966. updateIndexIfNeeded(project_id, doc_id, callback) {
  967. logger.log({ project_id, doc_id }, 'archiving old packs')
  968. return PackManager.getIndexWithKeys(doc_id, function (err, index) {
  969. if (err != null) {
  970. return callback(err)
  971. }
  972. if (index == null) {
  973. return PackManager.initialiseIndex(project_id, doc_id, callback)
  974. } else {
  975. return PackManager.updateIndex(project_id, doc_id, callback)
  976. }
  977. })
  978. },
  979. markPackAsChecked(project_id, doc_id, pack_id, callback) {
  980. logger.log({ project_id, doc_id, pack_id }, 'marking pack as checked')
  981. return db.docHistory.updateOne(
  982. { _id: pack_id },
  983. { $currentDate: { last_checked: true } },
  984. callback
  985. )
  986. },
  987. findUnarchivedPacks(project_id, doc_id, callback) {
  988. return PackManager.getIndex(doc_id, function (err, indexResult) {
  989. if (err != null) {
  990. return callback(err)
  991. }
  992. const indexPacks =
  993. (indexResult != null ? indexResult.packs : undefined) || []
  994. const unArchivedPacks = (() => {
  995. const result = []
  996. for (const pack of Array.from(indexPacks)) {
  997. if (pack.inS3 == null) {
  998. result.push(pack)
  999. }
  1000. }
  1001. return result
  1002. })()
  1003. if (unArchivedPacks.length) {
  1004. logger.log(
  1005. { project_id, doc_id, n: unArchivedPacks.length },
  1006. 'find unarchived packs'
  1007. )
  1008. }
  1009. return callback(null, unArchivedPacks)
  1010. })
  1011. },
  1012. // Archive locking flags
  1013. checkArchiveNotInProgress(project_id, doc_id, pack_id, callback) {
  1014. logger.log(
  1015. { project_id, doc_id, pack_id },
  1016. 'checking if archive in progress'
  1017. )
  1018. return PackManager.getPackFromIndex(
  1019. doc_id,
  1020. pack_id,
  1021. function (err, result) {
  1022. if (err != null) {
  1023. return callback(err)
  1024. }
  1025. if (result == null) {
  1026. return callback(new Error('pack not found in index'))
  1027. }
  1028. if (result.inS3) {
  1029. return callback(new Error('pack archiving already done'))
  1030. } else if (result.inS3 != null) {
  1031. return callback(new Error('pack archiving already in progress'))
  1032. } else {
  1033. return callback()
  1034. }
  1035. }
  1036. )
  1037. },
  1038. markPackAsArchiveInProgress(project_id, doc_id, pack_id, callback) {
  1039. logger.log(
  1040. { project_id, doc_id },
  1041. 'marking pack as archive in progress status'
  1042. )
  1043. return db.docHistoryIndex.findOneAndUpdate(
  1044. {
  1045. _id: ObjectId(doc_id.toString()),
  1046. packs: { $elemMatch: { _id: pack_id, inS3: { $exists: false } } },
  1047. },
  1048. { $set: { 'packs.$.inS3': false } },
  1049. { projection: { 'packs.$': 1 } },
  1050. function (err, result) {
  1051. if (err != null) {
  1052. return callback(err)
  1053. }
  1054. if (!result.value) {
  1055. return callback(new Error('archive is already in progress'))
  1056. }
  1057. logger.log(
  1058. { project_id, doc_id, pack_id },
  1059. 'marked as archive in progress'
  1060. )
  1061. return callback()
  1062. }
  1063. )
  1064. },
  1065. clearPackAsArchiveInProgress(project_id, doc_id, pack_id, callback) {
  1066. logger.log(
  1067. { project_id, doc_id, pack_id },
  1068. 'clearing as archive in progress'
  1069. )
  1070. return db.docHistoryIndex.updateOne(
  1071. {
  1072. _id: ObjectId(doc_id.toString()),
  1073. packs: { $elemMatch: { _id: pack_id, inS3: false } },
  1074. },
  1075. { $unset: { 'packs.$.inS3': true } },
  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.findOneAndUpdate(
  1082. {
  1083. _id: ObjectId(doc_id.toString()),
  1084. packs: { $elemMatch: { _id: pack_id, inS3: false } },
  1085. },
  1086. { $set: { 'packs.$.inS3': true } },
  1087. { projection: { 'packs.$': 1 } },
  1088. function (err, result) {
  1089. if (err != null) {
  1090. return callback(err)
  1091. }
  1092. if (!result.value) {
  1093. return callback(new Error('archive is not marked as progress'))
  1094. }
  1095. logger.log({ project_id, doc_id, pack_id }, 'marked as archived')
  1096. return callback()
  1097. }
  1098. )
  1099. },
  1100. setTTLOnArchivedPack(project_id, doc_id, pack_id, callback) {
  1101. return db.docHistory.updateOne(
  1102. { _id: pack_id },
  1103. { $set: { expiresAt: new Date(Date.now() + 1 * DAYS) } },
  1104. function (err) {
  1105. if (err) {
  1106. return callback(err)
  1107. }
  1108. logger.log({ project_id, doc_id, pack_id }, 'set expiry on pack')
  1109. return callback()
  1110. }
  1111. )
  1112. },
  1113. }
  1114. // _getOneDayInFutureWithRandomDelay: ->
  1115. // thirtyMins = 1000 * 60 * 30
  1116. // randomThirtyMinMax = Math.ceil(Math.random() * thirtyMins)
  1117. // return new Date(Date.now() + randomThirtyMinMax + 1*DAYS)