DocumentUpdaterManager.js 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150
  1. /* eslint-disable
  2. camelcase,
  3. */
  4. const request = require('request')
  5. const _ = require('underscore')
  6. const OError = require('@overleaf/o-error')
  7. const logger = require('@overleaf/logger')
  8. const settings = require('@overleaf/settings')
  9. const metrics = require('@overleaf/metrics')
  10. const {
  11. ClientRequestedMissingOpsError,
  12. DocumentUpdaterRequestFailedError,
  13. NullBytesInOpError,
  14. UpdateTooLargeError,
  15. } = require('./Errors')
  16. const rclient = require('@overleaf/redis-wrapper').createClient(
  17. settings.redis.documentupdater
  18. )
  19. const Keys = settings.redis.documentupdater.key_schema
  20. const DocumentUpdaterManager = {
  21. getDocument(project_id, doc_id, fromVersion, callback) {
  22. const timer = new metrics.Timer('get-document')
  23. const url = `${settings.apis.documentupdater.url}/project/${project_id}/doc/${doc_id}?fromVersion=${fromVersion}`
  24. logger.debug(
  25. { project_id, doc_id, fromVersion },
  26. 'getting doc from document updater'
  27. )
  28. request.get(url, function (err, res, body) {
  29. timer.done()
  30. if (err) {
  31. OError.tag(err, 'error getting doc from doc updater')
  32. return callback(err)
  33. }
  34. if (res.statusCode >= 200 && res.statusCode < 300) {
  35. logger.debug(
  36. { project_id, doc_id },
  37. 'got doc from document document updater'
  38. )
  39. try {
  40. body = JSON.parse(body)
  41. } catch (error) {
  42. OError.tag(error, 'error parsing doc updater response')
  43. return callback(error)
  44. }
  45. body = body || {}
  46. callback(null, body.lines, body.version, body.ranges, body.ops)
  47. } else if ([404, 422].includes(res.statusCode)) {
  48. callback(new ClientRequestedMissingOpsError(res.statusCode))
  49. } else {
  50. callback(
  51. new DocumentUpdaterRequestFailedError('getDocument', res.statusCode)
  52. )
  53. }
  54. })
  55. },
  56. checkDocument(project_id, doc_id, callback) {
  57. // in this call fromVersion = -1 means get document without docOps
  58. DocumentUpdaterManager.getDocument(project_id, doc_id, -1, callback)
  59. },
  60. flushProjectToMongoAndDelete(project_id, callback) {
  61. // this method is called when the last connected user leaves the project
  62. logger.debug({ project_id }, 'deleting project from document updater')
  63. const timer = new metrics.Timer('delete.mongo.project')
  64. // flush the project in the background when all users have left
  65. const url =
  66. `${settings.apis.documentupdater.url}/project/${project_id}?background=true` +
  67. (settings.shutDownInProgress ? '&shutdown=true' : '')
  68. request.del(url, function (err, res) {
  69. timer.done()
  70. if (err) {
  71. OError.tag(err, 'error deleting project from document updater')
  72. callback(err)
  73. } else if (res.statusCode >= 200 && res.statusCode < 300) {
  74. logger.debug({ project_id }, 'deleted project from document updater')
  75. callback(null)
  76. } else {
  77. callback(
  78. new DocumentUpdaterRequestFailedError(
  79. 'flushProjectToMongoAndDelete',
  80. res.statusCode
  81. )
  82. )
  83. }
  84. })
  85. },
  86. _getPendingUpdateListKey() {
  87. const shard = _.random(0, settings.pendingUpdateListShardCount - 1)
  88. if (shard === 0) {
  89. return 'pending-updates-list'
  90. } else {
  91. return `pending-updates-list-${shard}`
  92. }
  93. },
  94. queueChange(project_id, doc_id, change, callback) {
  95. const allowedKeys = [
  96. 'doc',
  97. 'op',
  98. 'v',
  99. 'dupIfSource',
  100. 'meta',
  101. 'lastV',
  102. 'hash',
  103. ]
  104. change = _.pick(change, allowedKeys)
  105. const jsonChange = JSON.stringify(change)
  106. if (jsonChange.indexOf('\u0000') !== -1) {
  107. // memory corruption check
  108. return callback(new NullBytesInOpError(jsonChange))
  109. }
  110. const updateSize = jsonChange.length
  111. if (updateSize > settings.maxUpdateSize) {
  112. return callback(new UpdateTooLargeError(updateSize))
  113. }
  114. // record metric for each update added to queue
  115. metrics.summary('redis.pendingUpdates', updateSize, { status: 'push' })
  116. const doc_key = `${project_id}:${doc_id}`
  117. // Push onto pendingUpdates for doc_id first, because once the doc updater
  118. // gets an entry on pending-updates-list, it starts processing.
  119. rclient.rpush(
  120. Keys.pendingUpdates({ doc_id }),
  121. jsonChange,
  122. function (error) {
  123. if (error) {
  124. error = new OError('error pushing update into redis').withCause(error)
  125. return callback(error)
  126. }
  127. const queueKey = DocumentUpdaterManager._getPendingUpdateListKey()
  128. rclient.rpush(queueKey, doc_key, function (error) {
  129. if (error) {
  130. error = new OError('error pushing doc_id into redis')
  131. .withInfo({ queueKey })
  132. .withCause(error)
  133. }
  134. callback(error)
  135. })
  136. }
  137. )
  138. },
  139. }
  140. module.exports = DocumentUpdaterManager