index.js 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239
  1. // @ts-check
  2. const crypto = require('node:crypto')
  3. const os = require('node:os')
  4. const { promisify } = require('node:util')
  5. // @ts-ignore - ioredis types may not be available in all contexts
  6. const Redis = require('ioredis')
  7. const {
  8. RedisHealthCheckTimedOut,
  9. RedisHealthCheckWriteError,
  10. RedisHealthCheckVerifyError,
  11. } = require('./Errors')
  12. const HEARTBEAT_TIMEOUT = 2000
  13. // generate unique values for health check
  14. const HOST = os.hostname()
  15. const PID = process.pid
  16. const RND = crypto.randomBytes(4).toString('hex')
  17. let COUNT = 0
  18. /**
  19. * @param {any} opts
  20. */
  21. function createClient(opts) {
  22. const standardOpts = Object.assign({}, opts)
  23. delete standardOpts.key_schema
  24. if (standardOpts.retry_max_delay == null) {
  25. standardOpts.retry_max_delay = 5000 // ms
  26. }
  27. if (standardOpts.endpoints) {
  28. throw new Error(
  29. '@overleaf/redis-wrapper: redis-sentinel is no longer supported'
  30. )
  31. }
  32. let client
  33. if (standardOpts.cluster) {
  34. delete standardOpts.cluster
  35. client = new Redis.Cluster(opts.cluster, standardOpts)
  36. } else {
  37. client = new Redis(standardOpts)
  38. }
  39. monkeyPatchIoRedisExec(client)
  40. /** @param {any} callback */
  41. client.healthCheck = callback => {
  42. if (callback) {
  43. // callback based invocation
  44. healthCheck(client).then(callback).catch(callback)
  45. } else {
  46. // Promise based invocation
  47. return healthCheck(client)
  48. }
  49. }
  50. return client
  51. }
  52. /**
  53. * @param {any} client
  54. */
  55. async function healthCheck(client) {
  56. // check the redis connection by storing and retrieving a unique key/value pair
  57. const uniqueToken = `host=${HOST}:pid=${PID}:random=${RND}:time=${Date.now()}:count=${COUNT++}`
  58. // o-error context
  59. const context = {
  60. uniqueToken,
  61. stage: 'add context for a timeout',
  62. }
  63. await runWithTimeout({
  64. runner: runCheck(client, uniqueToken, context),
  65. timeout: HEARTBEAT_TIMEOUT,
  66. context,
  67. })
  68. }
  69. /**
  70. * @param {any} client
  71. * @param {any} uniqueToken
  72. * @param {any} context
  73. */
  74. async function runCheck(client, uniqueToken, context) {
  75. const healthCheckKey = `_redis-wrapper:healthCheckKey:{${uniqueToken}}`
  76. const healthCheckValue = `_redis-wrapper:healthCheckValue:{${uniqueToken}}`
  77. // set the unique key/value pair
  78. context.stage = 'write'
  79. const writeAck = await client
  80. .set(healthCheckKey, healthCheckValue, 'EX', 60)
  81. .catch(
  82. /** @param {any} err */ err => {
  83. throw new RedisHealthCheckWriteError('write errored', context, err)
  84. }
  85. )
  86. if (writeAck !== 'OK') {
  87. context.writeAck = writeAck
  88. throw new RedisHealthCheckWriteError('write failed', context)
  89. }
  90. // check that we can retrieve the unique key/value pair
  91. context.stage = 'verify'
  92. const [roundTrippedHealthCheckValue, deleteAck] = await client
  93. .multi()
  94. .get(healthCheckKey)
  95. .del(healthCheckKey)
  96. .exec()
  97. .catch(
  98. /** @param {any} err */ err => {
  99. throw new RedisHealthCheckVerifyError(
  100. 'read/delete errored',
  101. context,
  102. err
  103. )
  104. }
  105. )
  106. if (roundTrippedHealthCheckValue !== healthCheckValue) {
  107. context.roundTrippedHealthCheckValue = roundTrippedHealthCheckValue
  108. throw new RedisHealthCheckVerifyError('read failed', context)
  109. }
  110. if (deleteAck !== 1) {
  111. context.deleteAck = deleteAck
  112. throw new RedisHealthCheckVerifyError('delete failed', context)
  113. }
  114. }
  115. /**
  116. * @param {any} result
  117. * @param {any} callback
  118. */
  119. function unwrapMultiResult(result, callback) {
  120. // ioredis exec returns a results like:
  121. // [ [null, 42], [null, "foo"] ]
  122. // where the first entries in each 2-tuple are
  123. // presumably errors for each individual command,
  124. // and the second entry is the result. We need to transform
  125. // this into the same result as the old redis driver:
  126. // [ 42, "foo" ]
  127. //
  128. // Basically reverse:
  129. // https://github.com/luin/ioredis/blob/v4.17.3/lib/utils/index.ts#L75-L92
  130. const filteredResult = []
  131. for (const [err, value] of result || []) {
  132. if (err) {
  133. return callback(err)
  134. } else {
  135. filteredResult.push(value)
  136. }
  137. }
  138. callback(null, filteredResult)
  139. }
  140. const unwrapMultiResultPromisified = promisify(unwrapMultiResult)
  141. /**
  142. * @param {any} client
  143. */
  144. function monkeyPatchIoRedisExec(client) {
  145. const _multi = client.multi
  146. client.multi = function () {
  147. const multi = _multi.apply(client, arguments)
  148. const _exec = multi.exec
  149. /** @param {any} callback */
  150. multi.exec = callback => {
  151. if (callback) {
  152. // callback based invocation
  153. _exec.call(
  154. multi,
  155. /**
  156. * @param {any} error
  157. * @param {any} result
  158. */
  159. (error, result) => {
  160. // The command can fail all-together due to syntax errors
  161. if (error) return callback(error)
  162. unwrapMultiResult(result, callback)
  163. }
  164. )
  165. } else {
  166. // Promise based invocation
  167. return _exec.call(multi).then(unwrapMultiResultPromisified)
  168. }
  169. }
  170. return multi
  171. }
  172. }
  173. /**
  174. * @param {{runner: any, timeout: any, context: any}} param0
  175. */
  176. async function runWithTimeout({ runner, timeout, context }) {
  177. /** @type {any} */
  178. let healthCheckDeadline
  179. await Promise.race([
  180. new Promise((resolve, reject) => {
  181. healthCheckDeadline = setTimeout(() => {
  182. // attach the timeout when hitting the timeout only
  183. context.timeout = timeout
  184. reject(new RedisHealthCheckTimedOut('timeout', context))
  185. }, timeout)
  186. }),
  187. runner.finally(() => clearTimeout(healthCheckDeadline)),
  188. ])
  189. }
  190. /**
  191. * Delete all data from the test Redis instance
  192. *
  193. * @param {Redis} rclient
  194. */
  195. async function cleanupTestRedis(rclient) {
  196. ensureTestRedis(rclient)
  197. await rclient.flushall()
  198. }
  199. /**
  200. * Checks that the Redis client points to a test database
  201. *
  202. * In tests, the Redis instance is on a host called redis_test
  203. *
  204. * @param {Redis} rclient
  205. */
  206. function ensureTestRedis(rclient) {
  207. const host = rclient.options.host
  208. const env = process.env.NODE_ENV
  209. if (host !== 'redis_test' || env !== 'test') {
  210. throw new Error(
  211. `Refusing to clear Redis instance '${host}' in environment '${env}'`
  212. )
  213. }
  214. }
  215. module.exports = {
  216. createClient,
  217. cleanupTestRedis,
  218. }