PackManager.js 32 KB

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