Nodejs redlock notes

redlock elementary analysis

Used in the project node-redlock This library , because Node The project is in Tencent taf Environment deployed , Has more than Docker Containers , The core requirement is to do a regular notification function for enterprise wechat robots .

Because there are many containers “ Contention concurrency ” The push message of , Therefore, a lock is needed to prevent repeated push .

Redlock The algorithm is Antirez In the single Redis High availability mode based on nodes .

The execution of the algorithm ：

1. Get the current time （ Millisecond unit ）
2. Try to start from N An example , Use the same key And unique value Get the lock , Direction Redis When requesting a lock , The client should Set a network connection and response timeout , This timeout should be less than the lock expiration time , This prevents the client from waiting
3. The client uses the current time minus the time to acquire the lock to get the time to acquire the lock . If and only if from more than half of Redis The node takes the lock , When the time used is less than the lock failure time , Lock is success
4. If you get the lock ,key The real effective time of is equal to the effective time minus the time used to acquire the lock , This is very important
5. If for some reason , Lock acquisition failed （ The lock is not obtained in more than half of the instances, or the lock removal time has exceeded the effective time ）, The client should be in all Redis Unlock the instance , No matter what Redis Whether the instance is locked successfully , Because the server response message may have been lost, but it actually succeeded , After all, one more release will not be a problem

Redlock The debate about whether the algorithm is safe

http://antirez.com/news/101

Read the whole article Redis Introduction to the official website , There is no perfect design , There is only one solution that fits the situation .

node-redlock The implementation of the

Plus source code 465 That's ok .

'use strict';

const util = require('util');
const crypto = require('crypto');
const Promise = require('bluebird');
const EventEmitter = require('events');

// constants
const lockScript = `
	-- Return 0 if an entry already exists.
	for i, key in ipairs(KEYS) do
		if redis.call("exists", key) == 1 then
			return 0
		end
	end
	-- Create an entry for each provided key.
	for i, key in ipairs(KEYS) do
		redis.call("set", key, ARGV[1], "PX", ARGV[2])
	end
	-- Return the number of entries added.
	return #KEYS
`;

const unlockScript = `
	local count = 0
	for i, key in ipairs(KEYS) do
		-- Only remove entries for *this* lock value.
		if redis.call("get", key) == ARGV[1] then
			redis.pcall("del", key)
			count = count + 1
		end
	end
	-- Return the number of entries removed.
	return count
`;

const extendScript = `
	-- Return 0 if an entry exists with a *different* lock value.
	for i, key in ipairs(KEYS) do
		if redis.call("get", key) ~= ARGV[1] then
			return 0
		end
	end
	-- Update the entry for each provided key.
	for i, key in ipairs(KEYS) do
		redis.call("set", key, ARGV[1], "PX", ARGV[2])
	end
	-- Return the number of entries updated.
	return #KEYS
`;

// defaults
const defaults = {
	driftFactor: 0.01,
	retryCount:  10,
	retryDelay:  200,
	retryJitter: 100
};





// LockError
// ---------
// This error is returned when there is an error locking a resource.
function LockError(message, attempts) {
	Error.call(this);
	Error.captureStackTrace(this, LockError);
	this.name = 'LockError';
	this.message = message || 'Failed to lock the resource.';
	this.attempts = attempts;
}

util.inherits(LockError, Error);






// Lock
// ----
// An object of this type is returned when a resource is successfully locked. It contains
// convenience methods `unlock` and `extend` which perform the associated Redlock method on
// itself.
function Lock(redlock, resource, value, expiration, attempts, attemptsRemaining) {
	this.redlock           = redlock;
	this.resource          = resource;
	this.value             = value;
	this.expiration        = expiration;
	this.attempts          = attempts;
	this.attemptsRemaining = attemptsRemaining;
}

Lock.prototype.unlock = function unlock(callback) {
	return this.redlock.unlock(this, callback);
};

Lock.prototype.extend = function extend(ttl, callback) {
	return this.redlock.extend(this, ttl, callback);
};

// Attach a reference to Lock, which allows the application to use instanceof
// to ensure type.
Redlock.Lock = Lock;


// Redlock
// -------
// A redlock object is instantiated with an array of at least one redis client and an optional
// `options` object. Properties of the Redlock object should NOT be changed after it is first
// used, as doing so could have unintended consequences for live locks.
function Redlock(clients, options) {
	// set default options
	options = options || {};
	this.driftFactor  = typeof options.driftFactor  === 'number' ? options.driftFactor : defaults.driftFactor;
	this.retryCount   = typeof options.retryCount   === 'number' ? options.retryCount  : defaults.retryCount;
	this.retryDelay   = typeof options.retryDelay   === 'number' ? options.retryDelay  : defaults.retryDelay;
	this.retryJitter  = typeof options.retryJitter  === 'number' ? options.retryJitter : defaults.retryJitter;
	this.lockScript   = typeof options.lockScript   === 'function' ? options.lockScript(lockScript) : lockScript;
	this.unlockScript = typeof options.unlockScript === 'function' ? options.unlockScript(unlockScript) : unlockScript;
	this.extendScript = typeof options.extendScript === 'function' ? options.extendScript(extendScript) : extendScript;
	// set the redis servers from additional arguments
	this.servers = clients;
	if(this.servers.length === 0)
		throw new Error('Redlock must be instantiated with at least one redis server.');

	this.scripts = {
		lockScript: { value: this.lockScript, hash: this._hashScript(this.lockScript) },
		unlockScript: { value: this.unlockScript, hash: this._hashScript(this.unlockScript) },
		extendScript: { value: this.extendScript, hash: this._hashScript(this.extendScript) },
	};
}

// Inherit all the EventEmitter methods, like `on`, and `off`
util.inherits(Redlock, EventEmitter);


// Attach a reference to LockError per issue #7, which allows the application to use instanceof
// to destinguish between error types.
Redlock.LockError = LockError;

// quit
// ----
// This method runs `.quit()` on all client connections.

Redlock.prototype.quit = function quit(callback) {

	// quit all clients
	return Promise.map(this.servers, function(client) {
		return client.quit();
	})

	// optionally run callback
	.nodeify(callback); // bluebird.js  A unique way of writing , promise  The callback   The first parameter is the error message , If successful, then  null,  The second parameter is success information 
};


// lock  Lock a given resource 
// ----
// This method locks a resource using the redlock algorithm.
//
// ```js
// redlock.lock(
//   'some-resource',       // the resource to lock
//   2000,                  // ttl in ms
//   function(err, lock) {  // callback function (optional)
//     ...
//   }
// )
// ```
Redlock.prototype.acquire =
Redlock.prototype.lock = function lock(resource, ttl, callback) {
	return this._lock(resource, null, ttl, {}, callback);
};

// lockWithOptions
// ---------------
// This method locks a resource and overwrites some of the options
// ```js
// redlock.lockWithOptions(
//   'some-resource',                    // the resource to lock
//   2000,                               // ttl in ms
//   { retryCount: 1, retryDelay: 100 }, // additional options
//   function(err, lock) {               // callback function (optional)
//     ...
//   }
// )
// ```
Redlock.prototype.acquireWithOptions =
Redlock.prototype.lockWithOptions = function lock(resource, ttl, options, callback) {
	return this._lock(resource, null, ttl, options, callback);
};

// lock
// ----
// This method locks a resource using the redlock algorithm,
// and returns a bluebird disposer.
//
// ```js
// using(
//   redlock.disposer(
//     'some-resource',       // the resource to lock
//     2000                   // ttl in ms
//   ),
//   function(lock) {
//     ...
//   }
// );
// ```
Redlock.prototype.disposer = function disposer(resource, ttl, errorHandler) {
	errorHandler = errorHandler || function(err) {};
	return this._lock(resource, null, ttl, {}).disposer(function(lock){
		return lock.unlock().catch(errorHandler);
	});
};


// unlock
// ------
// This method unlocks the provided lock from all servers still persisting it. It will fail
// with an error if it is unable to release the lock on a quorum of nodes, but will make no
// attempt to restore the lock on nodes that failed to release. It is safe to re-attempt an
// unlock or to ignore the error, as the lock will automatically expire after its timeout.
Redlock.prototype.release =
Redlock.prototype.unlock = function unlock(lock, callback) {
	const self = this;

	// array of locked resources
	const resource = Array.isArray(lock.resource)
		? lock.resource
		: [lock.resource];
	
	// immediately invalidate the lock
	lock.expiration = 0;

	return new Promise(function(resolve, reject) {

		// the number of votes needed for consensus
		const quorum = Math.floor(self.servers.length / 2) + 1;

		// the number of servers which have agreed to release this lock
		let votes = 0;

		// the number of async redis calls still waiting to finish
		let waiting = self.servers.length;

		// release the lock on each server
		self.servers.forEach(function(server){
			return self._executeScript(server, 'unlockScript', [
					resource.length,
					...resource,
					lock.value
			], loop);
		});

		function loop(err, response) {
			if(err) self.emit('clientError', err);

			// - If the response is less than the resource length, than one or
			//   more resources failed to unlock:
			//   - It may have been re-acquired by another process;
			//   - It may hava already been manually released;
			//   - It may have expired;

			if(response === resource.length || response === '' + resource.length)
				votes++;

			if(waiting-- > 1) return;

			// SUCCESS: there is concensus and the lock is released
			if(votes >= quorum)
				return resolve();

			// FAILURE: the lock could not be released
			return reject(new LockError('Unable to fully release the lock on resource "' + lock.resource + '".'));
		}
	})

	// optionally run callback
	.nodeify(callback);
};


// extend
// ------
// This method extends a valid lock by the provided `ttl`.
Redlock.prototype.extend = function extend(lock, ttl, callback) {
	const self = this;

	// the lock has expired
	if(lock.expiration < Date.now())
		return Promise.reject(new LockError('Cannot extend lock on resource "' + lock.resource + '" because the lock has already expired.', 0)).nodeify(callback);

	// extend the lock
	return self._lock(lock.resource, lock.value, ttl, {})

	// modify and return the original lock object
	.then(function(extension){
		lock.value      = extension.value;
		lock.expiration = extension.expiration;
		return lock;
	})

	// optionally run callback
	.nodeify(callback);
};


// _lock
// -----
// This method locks a resource using the redlock algorithm.
//
// ###Creating New Locks:
//
// ```js
// redlock._lock(
//   'some-resource',       // the resource to lock
//   null,                  // no original lock value
//   2000,                  // ttl in ms
//   {},                    // option overrides {retryCount, retryDelay}
//   function(err, lock) {  // callback function (optional)
//     ...
//   }
// )
// ```
//
// ###Extending Existing Locks:
//
// ```js
// redlock._lock(
//   'some-resource',       // the resource to lock
//   'dkkk18g4gy39dx6r',    // the value of the original lock
//   2000,                  // ttl in ms
//   {},                    // option overrides {retryCount, retryDelay}
//   function(err, lock) {  // callback function (optional)
//     ...
//   }
// )
// ```
Redlock.prototype._lock = function _lock(resource, value, ttl, options, callback) {
	const self = this;

	// backwards compatibility with previous method signature: _lock(resource, value, ttl, callback)
	if (typeof options === 'function' && typeof callback === 'undefined') {
		callback = options;
		options = {};
	}

	// array of locked resources
	resource = Array.isArray(resource) ? resource : [resource];

	return new Promise(function(resolve, reject) {
		let request;

		// the number of times we have attempted this lock
		let attempts = 0;

		// create a new lock
		if(value === null) {
			value = self._random();
			request = function(server, loop){
				//  perform  lockScript  Script 
				return self._executeScript(server, 'lockScript', [
						resource.length,
						...resource,
						value,
						ttl
				], loop);
			};
		}

		// extend an existing lock
		else {
			request = function(server, loop){
				//  perform  extendScript  Script 
				return self._executeScript(server, 'extendScript', [
						resource.length,
						...resource,
						value,
						ttl
				], loop);
			};
		}

		function attempt(){
			attempts++;

			let retryCount = options.retryCount || self.retryCount;
			let retryDelay = options.retryDelay || self.retryDelay;

			// the time when this attempt started
			const start = Date.now(); //  Starting time 

			// the number of votes needed for consensus
			const quorum = Math.floor(self.servers.length / 2) + 1; //  Only more than half of the instances can consider resources lockable 

			// the number of servers which have agreed to this lock
			let votes = 0; //  The instance that has obtained the lock currently 

			// the number of async redis calls still waiting to finish
			let waiting = self.servers.length; //  Instance of lock not acquired 

			function loop(err, response) {
				if(err) self.emit('clientError', err);
				if(response === resource.length || response === '' + resource.length) votes++;
				if(waiting-- > 1) return;
				
				// Add 2 milliseconds to the drift to account for Redis expires precision, which is 1 ms,
				// plus the configured allowable drift factor
				const drift = Math.round(self.driftFactor * ttl) + 2;
				const lock = new Lock(self, resource, value, start + ttl - drift, attempts, retryCount - attempts);
				
				// SUCCESS: there is concensus and the lock is not expired
				if(votes >= quorum && lock.expiration > Date.now())
					return resolve(lock);


				// remove this lock from servers that voted for it
				return lock.unlock(function(){

					// RETRY
					if(retryCount === -1 || attempts <= retryCount)
						return setTimeout(attempt, Math.max(0, retryDelay + Math.floor((Math.random() * 2 - 1) * self.retryJitter)));

					// FAILED
					return reject(new LockError('Exceeded ' + retryCount + ' attempts to lock the resource "' + resource + '".', attempts));
				});
			}

			//  To each  redis  Instance makes a request to try to lock , More than half of the operations are completed 
			return self.servers.forEach(function(server){
				return request(server, loop);
			});
		}

		return attempt();
	})

	// optionally run callback
	.nodeify(callback);
};

//  In order to meet the algorithm requirements , Create random  value
Redlock.prototype._random = function _random(){
	return crypto.randomBytes(16).toString('hex');
};

Redlock.prototype._executeScript = function(server, name, args, callback) {
	const script = this.scripts[name];

	// server  Is to initialize the incoming  redis  example ,Redis Evalsha  The command is based on the given  sha1  Check code , Executing scripts cached in the server 
	return server.evalsha(script.hash, args, (err, result) => {
		if(err !== null && err.message.startsWith("NOSCRIPT")) {
			// Script is not loaded yet, call eval and it will populate it in redis lua scripts cache
			args.unshift(script.value);
			//  If there is no cached script ,Redis Eval  Command to use  Lua  The interpreter executes the script 
			return server.eval(args, callback);
		}

		return callback(err, result);
	});
}

//  establish  hash, Used to execute scripts cached in the server 
Redlock.prototype._hashScript = function(value) {
	return crypto.createHash('sha1').update(value).digest('hex');
}

module.exports = Redlock;