mirror of
https://github.com/overleaf/overleaf.git
synced 2024-11-21 20:47:08 -05:00
Merge pull request #7835 from overleaf/jpa-redis-locker-package
[misc] extract document-updaters LockManager into redis-wrapper package GitOrigin-RevId: 5775ab9df7afb614dfc5f502df526c8735f84a91
This commit is contained in:
Jakob Ackermann
Copybot
parent
9b320f5286
commit
cd11db456d
5 changed files with 198 additions and 175 deletions
175
libraries/redis-wrapper/RedisLocker.js
Normal file
175
libraries/redis-wrapper/RedisLocker.js
Normal file
|
|
@ -0,0 +1,175 @@
|
|||
const metrics = require('@overleaf/metrics')
|
||||
const logger = require('@overleaf/logger')
|
||||
const os = require('os')
|
||||
const crypto = require('crypto')
|
||||
|
||||
const HOST = os.hostname()
|
||||
const PID = process.pid
|
||||
const RND = crypto.randomBytes(4).toString('hex')
|
||||
let COUNT = 0
|
||||
|
||||
const MAX_REDIS_REQUEST_LENGTH = 5000 // 5 seconds
|
||||
|
||||
const UNLOCK_SCRIPT =
|
||||
'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end'
|
||||
|
||||
module.exports = class RedisLocker {
|
||||
/**
|
||||
* @param rclient initialized ioredis client
|
||||
* @param getKey compose the redis key based on the passed id
|
||||
* @param wrapTimeoutError assign the id to a designated field on the error
|
||||
* @param metricsPrefix prefix all the metrics with the given prefix
|
||||
*
|
||||
* @example ```
|
||||
* const lock = new RedisLocker({
|
||||
* rclient,
|
||||
* getKey(userId) { return `blocking:{userId}` },
|
||||
* wrapTimeoutError(err, userId) { err.userId = userId; return err },
|
||||
* metricsPrefix: 'user',
|
||||
* })
|
||||
*
|
||||
* lock.getLock(user._id, (err, value) => {
|
||||
* if (err) return callback(err)
|
||||
* // do work
|
||||
* lock.releaseLock(user._id, callback)
|
||||
* }
|
||||
* ```
|
||||
*/
|
||||
constructor({ rclient, getKey, wrapTimeoutError, metricsPrefix }) {
|
||||
this.rclient = rclient
|
||||
this.getKey = getKey
|
||||
this.wrapTimeoutError = wrapTimeoutError
|
||||
this.metricsPrefix = metricsPrefix
|
||||
|
||||
this.LOCK_TEST_INTERVAL = 50 // 50ms between each test of the lock
|
||||
this.MAX_TEST_INTERVAL = 1000 // back off to 1s between each test of the lock
|
||||
this.MAX_LOCK_WAIT_TIME = 10000 // 10s maximum time to spend trying to get the lock
|
||||
this.LOCK_TTL = 30 // seconds. Time until lock auto expires in redis.
|
||||
|
||||
// read-only copy for unit tests
|
||||
this.unlockScript = UNLOCK_SCRIPT
|
||||
}
|
||||
|
||||
// Use a signed lock value as described in
|
||||
// https://redis.io/docs/reference/patterns/distributed-locks/#correct-implementation-with-a-single-instance
|
||||
// to prevent accidental unlocking by multiple processes
|
||||
randomLock() {
|
||||
const time = Date.now()
|
||||
return `locked:host=${HOST}:pid=${PID}:random=${RND}:time=${time}:count=${COUNT++}`
|
||||
}
|
||||
|
||||
tryLock(id, callback) {
|
||||
if (callback == null) {
|
||||
callback = function () {}
|
||||
}
|
||||
const lockValue = this.randomLock()
|
||||
const key = this.getKey(id)
|
||||
const startTime = Date.now()
|
||||
return this.rclient.set(
|
||||
key,
|
||||
lockValue,
|
||||
'EX',
|
||||
this.LOCK_TTL,
|
||||
'NX',
|
||||
(err, gotLock) => {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
}
|
||||
if (gotLock === 'OK') {
|
||||
metrics.inc(this.metricsPrefix + '-not-blocking')
|
||||
const timeTaken = Date.now() - startTime
|
||||
if (timeTaken > MAX_REDIS_REQUEST_LENGTH) {
|
||||
// took too long, so try to free the lock
|
||||
return this.releaseLock(id, lockValue, function (err, result) {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
} // error freeing lock
|
||||
return callback(null, false)
|
||||
}) // tell caller they didn't get the lock
|
||||
} else {
|
||||
return callback(null, true, lockValue)
|
||||
}
|
||||
} else {
|
||||
metrics.inc(this.metricsPrefix + '-blocking')
|
||||
return callback(null, false)
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
getLock(id, callback) {
|
||||
if (callback == null) {
|
||||
callback = function () {}
|
||||
}
|
||||
const startTime = Date.now()
|
||||
let testInterval = this.LOCK_TEST_INTERVAL
|
||||
const attempt = () => {
|
||||
if (Date.now() - startTime > this.MAX_LOCK_WAIT_TIME) {
|
||||
const e = this.wrapTimeoutError(new Error('Timeout'), id)
|
||||
return callback(e)
|
||||
}
|
||||
|
||||
return this.tryLock(id, (error, gotLock, lockValue) => {
|
||||
if (error != null) {
|
||||
return callback(error)
|
||||
}
|
||||
if (gotLock) {
|
||||
return callback(null, lockValue)
|
||||
} else {
|
||||
setTimeout(attempt, testInterval)
|
||||
// back off when the lock is taken to avoid overloading
|
||||
return (testInterval = Math.min(
|
||||
testInterval * 2,
|
||||
this.MAX_TEST_INTERVAL
|
||||
))
|
||||
}
|
||||
})
|
||||
}
|
||||
attempt()
|
||||
}
|
||||
|
||||
checkLock(id, callback) {
|
||||
if (callback == null) {
|
||||
callback = function () {}
|
||||
}
|
||||
const key = this.getKey(id)
|
||||
return this.rclient.exists(key, (err, exists) => {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
}
|
||||
exists = parseInt(exists)
|
||||
if (exists === 1) {
|
||||
metrics.inc(this.metricsPrefix + '-blocking')
|
||||
return callback(null, false)
|
||||
} else {
|
||||
metrics.inc(this.metricsPrefix + '-not-blocking')
|
||||
return callback(null, true)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
releaseLock(id, lockValue, callback) {
|
||||
const key = this.getKey(id)
|
||||
return this.rclient.eval(
|
||||
UNLOCK_SCRIPT,
|
||||
1,
|
||||
key,
|
||||
lockValue,
|
||||
(err, result) => {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
} else if (result != null && result !== 1) {
|
||||
// successful unlock should release exactly one key
|
||||
logger.error(
|
||||
{ id, key, lockValue, redis_err: err, redis_result: result },
|
||||
'unlocking error'
|
||||
)
|
||||
metrics.inc(this.metricsPrefix + '-unlock-error')
|
||||
return callback(new Error('tried to release timed out lock'))
|
||||
} else {
|
||||
return callback(null, result)
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -5,7 +5,8 @@
|
|||
"main": "index.js",
|
||||
"files": [
|
||||
"index.js",
|
||||
"Errors.js"
|
||||
"Errors.js",
|
||||
"RedisLocker.js"
|
||||
],
|
||||
"author": "Overleaf (https://www.overleaf.com)",
|
||||
"repository": "github:overleaf/redis-wrapper",
|
||||
|
|
@ -20,6 +21,7 @@
|
|||
},
|
||||
"peerDependencies": {
|
||||
"@overleaf/logger": "*",
|
||||
"@overleaf/metrics": "*",
|
||||
"@overleaf/o-error": "^3.4.0"
|
||||
},
|
||||
"dependencies": {
|
||||
|
|
|
|||
1
package-lock.json
generated
1
package-lock.json
generated
|
|
@ -930,6 +930,7 @@
|
|||
},
|
||||
"peerDependencies": {
|
||||
"@overleaf/logger": "*",
|
||||
"@overleaf/metrics": "*",
|
||||
"@overleaf/o-error": "^3.4.0"
|
||||
}
|
||||
},
|
||||
|
|
|
|||
|
|
@ -1,176 +1,17 @@
|
|||
/* eslint-disable
|
||||
camelcase,
|
||||
no-return-assign,
|
||||
*/
|
||||
// TODO: This file was created by bulk-decaffeinate.
|
||||
// Fix any style issues and re-enable lint.
|
||||
/*
|
||||
* decaffeinate suggestions:
|
||||
* DS102: Remove unnecessary code created because of implicit returns
|
||||
* DS207: Consider shorter variations of null checks
|
||||
* Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md
|
||||
*/
|
||||
let LockManager
|
||||
const metrics = require('./Metrics')
|
||||
const Settings = require('@overleaf/settings')
|
||||
const redis = require('@overleaf/redis-wrapper')
|
||||
const rclient = redis.createClient(Settings.redis.lock)
|
||||
const keys = Settings.redis.lock.key_schema
|
||||
const logger = require('@overleaf/logger')
|
||||
const os = require('os')
|
||||
const crypto = require('crypto')
|
||||
const RedisLocker = require('@overleaf/redis-wrapper/RedisLocker')
|
||||
|
||||
const Profiler = require('./Profiler')
|
||||
|
||||
const HOST = os.hostname()
|
||||
const PID = process.pid
|
||||
const RND = crypto.randomBytes(4).toString('hex')
|
||||
let COUNT = 0
|
||||
|
||||
const MAX_REDIS_REQUEST_LENGTH = 5000 // 5 seconds
|
||||
|
||||
module.exports = LockManager = {
|
||||
LOCK_TEST_INTERVAL: 50, // 50ms between each test of the lock
|
||||
MAX_TEST_INTERVAL: 1000, // back off to 1s between each test of the lock
|
||||
MAX_LOCK_WAIT_TIME: 10000, // 10s maximum time to spend trying to get the lock
|
||||
LOCK_TTL: 30, // seconds. Time until lock auto expires in redis.
|
||||
|
||||
// Use a signed lock value as described in
|
||||
// http://redis.io/topics/distlock#correct-implementation-with-a-single-instance
|
||||
// to prevent accidental unlocking by multiple processes
|
||||
randomLock() {
|
||||
const time = Date.now()
|
||||
return `locked:host=${HOST}:pid=${PID}:random=${RND}:time=${time}:count=${COUNT++}`
|
||||
module.exports = new RedisLocker({
|
||||
rclient,
|
||||
getKey(docId) {
|
||||
return keys.blockingKey({ doc_id: docId })
|
||||
},
|
||||
|
||||
unlockScript:
|
||||
'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end',
|
||||
|
||||
tryLock(doc_id, callback) {
|
||||
if (callback == null) {
|
||||
callback = function () {}
|
||||
}
|
||||
const lockValue = LockManager.randomLock()
|
||||
const key = keys.blockingKey({ doc_id })
|
||||
const profile = new Profiler('tryLock', { doc_id, key, lockValue })
|
||||
return rclient.set(
|
||||
key,
|
||||
lockValue,
|
||||
'EX',
|
||||
this.LOCK_TTL,
|
||||
'NX',
|
||||
function (err, gotLock) {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
}
|
||||
if (gotLock === 'OK') {
|
||||
metrics.inc('doc-not-blocking')
|
||||
const timeTaken = profile.log('got lock').end()
|
||||
if (timeTaken > MAX_REDIS_REQUEST_LENGTH) {
|
||||
// took too long, so try to free the lock
|
||||
return LockManager.releaseLock(
|
||||
doc_id,
|
||||
lockValue,
|
||||
function (err, result) {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
} // error freeing lock
|
||||
return callback(null, false)
|
||||
}
|
||||
) // tell caller they didn't get the lock
|
||||
} else {
|
||||
return callback(null, true, lockValue)
|
||||
}
|
||||
} else {
|
||||
metrics.inc('doc-blocking')
|
||||
profile.log('doc is locked').end()
|
||||
return callback(null, false)
|
||||
}
|
||||
}
|
||||
)
|
||||
wrapTimeoutError(err, docId) {
|
||||
err.doc_id = docId
|
||||
return err
|
||||
},
|
||||
|
||||
getLock(doc_id, callback) {
|
||||
let attempt
|
||||
if (callback == null) {
|
||||
callback = function () {}
|
||||
}
|
||||
const startTime = Date.now()
|
||||
let testInterval = LockManager.LOCK_TEST_INTERVAL
|
||||
const profile = new Profiler('getLock', { doc_id })
|
||||
return (attempt = function () {
|
||||
if (Date.now() - startTime > LockManager.MAX_LOCK_WAIT_TIME) {
|
||||
const e = new Error('Timeout')
|
||||
e.doc_id = doc_id
|
||||
profile.log('timeout').end()
|
||||
return callback(e)
|
||||
}
|
||||
|
||||
return LockManager.tryLock(doc_id, function (error, gotLock, lockValue) {
|
||||
if (error != null) {
|
||||
return callback(error)
|
||||
}
|
||||
profile.log('tryLock')
|
||||
if (gotLock) {
|
||||
profile.end()
|
||||
return callback(null, lockValue)
|
||||
} else {
|
||||
setTimeout(attempt, testInterval)
|
||||
// back off when the lock is taken to avoid overloading
|
||||
return (testInterval = Math.min(
|
||||
testInterval * 2,
|
||||
LockManager.MAX_TEST_INTERVAL
|
||||
))
|
||||
}
|
||||
})
|
||||
})()
|
||||
},
|
||||
|
||||
checkLock(doc_id, callback) {
|
||||
if (callback == null) {
|
||||
callback = function () {}
|
||||
}
|
||||
const key = keys.blockingKey({ doc_id })
|
||||
return rclient.exists(key, function (err, exists) {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
}
|
||||
exists = parseInt(exists)
|
||||
if (exists === 1) {
|
||||
metrics.inc('doc-blocking')
|
||||
return callback(null, false)
|
||||
} else {
|
||||
metrics.inc('doc-not-blocking')
|
||||
return callback(null, true)
|
||||
}
|
||||
})
|
||||
},
|
||||
|
||||
releaseLock(doc_id, lockValue, callback) {
|
||||
const key = keys.blockingKey({ doc_id })
|
||||
const profile = new Profiler('releaseLock', { doc_id, key, lockValue })
|
||||
return rclient.eval(
|
||||
LockManager.unlockScript,
|
||||
1,
|
||||
key,
|
||||
lockValue,
|
||||
function (err, result) {
|
||||
if (err != null) {
|
||||
return callback(err)
|
||||
} else if (result != null && result !== 1) {
|
||||
// successful unlock should release exactly one key
|
||||
profile.log('unlockScript:expired-lock').end()
|
||||
logger.error(
|
||||
{ doc_id, key, lockValue, redis_err: err, redis_result: result },
|
||||
'unlocking error'
|
||||
)
|
||||
metrics.inc('unlock-error')
|
||||
return callback(new Error('tried to release timed out lock'))
|
||||
} else {
|
||||
profile.log('unlockScript:ok').end()
|
||||
return callback(null, result)
|
||||
}
|
||||
}
|
||||
)
|
||||
},
|
||||
}
|
||||
metricsPrefix: 'doc',
|
||||
})
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@
|
|||
const sinon = require('sinon')
|
||||
const modulePath = '../../../../app/js/LockManager.js'
|
||||
const SandboxedModule = require('sandboxed-module')
|
||||
const tk = require('timekeeper')
|
||||
|
||||
describe('LockManager - trying the lock', function () {
|
||||
beforeEach(function () {
|
||||
|
|
@ -95,14 +96,17 @@ describe('LockManager - trying the lock', function () {
|
|||
|
||||
return describe('when it takes a long time for redis to set the lock', function () {
|
||||
beforeEach(function () {
|
||||
this.Profiler.prototype.end = () => 7000 // take a long time
|
||||
this.Profiler.prototype.log = sinon
|
||||
.stub()
|
||||
.returns({ end: this.Profiler.prototype.end })
|
||||
tk.freeze(Date.now())
|
||||
this.lockValue = 'mock-lock-value'
|
||||
this.LockManager.randomLock = sinon.stub().returns(this.lockValue)
|
||||
this.LockManager.releaseLock = sinon.stub().callsArgWith(2, null)
|
||||
return this.set.callsArgWith(5, null, 'OK')
|
||||
this.set.callsFake((_key, _v, _ex, _ttl, _nx, cb) => {
|
||||
tk.freeze(Date.now() + 7000)
|
||||
cb(null, 'OK')
|
||||
})
|
||||
})
|
||||
after(function () {
|
||||
tk.reset()
|
||||
})
|
||||
|
||||
describe('in all cases', function () {
|
||||
|
|
|
|||
Loading…
Reference in a new issue