PackManager.js 34 KB

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