PackManager.js 32 KB

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