import { randomBytes , createHash } from "crypto" ; import { EventEmitter } from "events" ; import { AbortController as PolyfillAbortController } from "node-abort-controller" ; import { Redis as IORedisClient , Cluster as IORedisCluster } from "ioredis" ; type Client = IORedisClient | IORedisCluster ; const ACQUIRE_SCRIPT = `
  -- 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 EXTEND_SCRIPT = `
  -- 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
` ; const RELEASE_SCRIPT = `
  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
` ; export type ClientExecutionResult = { client : Client ; vote : "for" ; value : number ; } | { client : Client ; vote : "against" ; error : Error ; } ; export type ExecutionStats = { readonly membershipSize : number ; readonly quorumSize : number ; readonly votesFor : Set < Client > ; readonly votesAgainst : Map < Client , Error > ; } ; export type ExecutionResult = { attempts : ReadonlyArray < Promise < ExecutionStats > > ; } ; /**
 *
 */ export interface Settings { readonly driftFactor : number ; readonly retryCount : number ; readonly retryDelay : number ; readonly retryJitter : number ; readonly automaticExtensionThreshold : number ; } const defaultSettings = { driftFactor : 0.01 , retryCount : 10 , retryDelay : 200 , retryJitter : 100 , automaticExtensionThreshold : 500 , } ; Object . freeze ( defaultSettings ) ; export class ResourceLockedError extends Error { constructor ( public readonly message ) { super ( ) ; this . name = "ResourceLockedError" ; } } export class ExecutionError extends Error { constructor ( public readonly message , public readonly attempts ) { super ( ) ; this . name = "ExecutionError" ; } } export class Lock { constructor ( public readonly redlock , public readonly resources , public readonly value , public readonly attempts , public expiration ) { } async release ( ) { return this . redlock . release ( this ) ; } async extend ( duration ) { return this . redlock . extend ( this , duration ) ; } } type RedlockAbortSignal = AbortSignal & { error ? : Error ; } ; /**
 * 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.
 */ export default class Redlock extends EventEmitter { public readonly clients : Set < Client > ; public readonly settings : Settings ; public readonly scripts : { readonly acquireScript : { value : string ; hash : string ; } ; readonly extendScript : { value : string ; hash : string ; } ; readonly releaseScript : { value : string ; hash : string ; } ; } ; public constructor ( clients , settings = { } , scripts = { } ) { super ( ) ; this . on ( "error" , ( ) => { } ) ; this . clients = new Set ( clients ) ; if ( this . clients . size === 0 ) { throw new Error ( "Redlock must be instantiated with at least one redis client." ) ; } this . settings = { driftFactor : typeof settings . driftFactor === "number" ? settings . driftFactor : defaultSettings . driftFactor , retryCount : typeof settings . retryCount === "number" ? settings . retryCount : defaultSettings . retryCount , retryDelay : typeof settings . retryDelay === "number" ? settings . retryDelay : defaultSettings . retryDelay , retryJitter : typeof settings . retryJitter === "number" ? settings . retryJitter : defaultSettings . retryJitter , automaticExtensionThreshold : typeof settings . automaticExtensionThreshold === "number" ? settings . automaticExtensionThreshold : defaultSettings . automaticExtensionThreshold , } ; const acquireScript = typeof scripts . acquireScript === "function" ? scripts . acquireScript ( ACQUIRE_SCRIPT ) : ACQUIRE_SCRIPT ; const extendScript = typeof scripts . extendScript === "function" ? scripts . extendScript ( EXTEND_SCRIPT ) : EXTEND_SCRIPT ; const releaseScript = typeof scripts . releaseScript === "function" ? scripts . releaseScript ( RELEASE_SCRIPT ) : RELEASE_SCRIPT ; this . scripts = { acquireScript : { value : acquireScript , hash : this . _hash ( acquireScript ) , } , extendScript : { value : extendScript , hash : this . _hash ( extendScript ) , } , releaseScript : { value : releaseScript , hash : this . _hash ( releaseScript ) , } , } ; } /**
     * Generate a sha1 hash compatible with redis evalsha.
     */ private _hash ( value ) { return createHash ( "sha1" ) . update ( value ) . digest ( "hex" ) ; } /**
     * Generate a cryptographically random string.
     */ private _random ( ) { return randomBytes ( 16 ) . toString ( "hex" ) ; } /**
     * This method runs `.quit()` on all client connections.
     */ public async quit ( ) { const results = [ ] ; for ( const client of this . clients ) { results . push ( client . quit ( ) ) ; } await Promise . all ( results ) ; } /**
     * This method acquires a locks on the resources for the duration specified by
     * the `duration`.
     */ public async acquire ( resources , duration , settings ? ) { const start = Date . now ( ) ; const value = this . _random ( ) ; try { const { attempts } = await this . _execute ( this . scripts . acquireScript , resources , [ value , duration ] , settings ) ; const drift = Math . round ( ( settings ?. driftFactor ?? this . settings . driftFactor ) * duration ) + 2 ; return new Lock ( this , resources , value , attempts , start + duration - drift ) ; } catch ( error ) { await this . _execute ( this . scripts . releaseScript , resources , [ value ] , { retryCount : 0 , } ) . catch ( ( ) => { } ) ; throw error ; } } /**
     * 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 in the case of a
     * failure to release. It is safe to re-attempt a release or to ignore the
     * error, as the lock will automatically expire after its timeout.
     */ public async release ( lock , settings ? ) { lock . expiration = 0 ; return this . _execute ( this . scripts . releaseScript , lock . resources , [ lock . value ] , settings ) ; } /**
     * This method extends a valid lock by the provided `duration`.
     */ public async extend ( existing , duration , settings ? ) { const start = Date . now ( ) ; if ( existing . expiration < Date . now ( ) ) { throw new ExecutionError ( "Cannot extend an already-expired lock." , [ ] ) ; } const { attempts } = await this . _execute ( this . scripts . extendScript , existing . resources , [ existing . value , duration ] , settings ) ; existing . expiration = 0 ; const drift = Math . round ( ( settings ?. driftFactor ?? this . settings . driftFactor ) * duration ) + 2 ; const replacement = new Lock ( this , existing . resources , existing . value , attempts , start + duration - drift ) ; return replacement ; } /**
     * Execute a script on all clients. The resulting promise is resolved or
     * rejected as soon as this quorum is reached; the resolution or rejection
     * will contains a `stats` property that is resolved once all votes are in.
     */ private async _execute ( script , keys , args , _settings ? ) { const settings = _settings ? { ... this . settings , ... _settings , } : this . settings ; const maxAttempts = settings . retryCount + 1 ; const attempts = [ ] ; while ( true ) { const { vote , stats } = await this . _attemptOperation ( script , keys , args ) ; attempts . push ( stats ) ; if ( vote === "for" ) { return { attempts } ; } if ( attempts . length < maxAttempts ) { await new Promise ( ( resolve ) => { setTimeout ( resolve , Math . max ( 0 , settings . retryDelay + Math . floor ( ( Math . random ( ) * 2 - 1 ) * settings . retryJitter ) ) , undefined ) ; } ) ; } else { throw new ExecutionError ( "The operation was unable to acheive a quorum during its retry window." , attempts ) ; } } } private async _attemptOperation ( script , keys , args ) { return await new Promise ( ( resolve ) => { const clientResults = [ ] ; for ( const client of this . clients ) { clientResults . push ( this . _attemptOperationOnClient ( client , script , keys , args ) ) ; } const stats = { membershipSize : clientResults . length , quorumSize : Math . floor ( clientResults . length / 2 ) + 1 , votesFor : new Set < Client > ( ) , votesAgainst : new Map < Client , Error > ( ) , } ; let done ; const statsPromise = new Promise < typeof stats > ( ( resolve ) => { done = ( ) => resolve ( stats ) ; } ) ; const onResultResolve = ( clientResult ) : void => { switch ( clientResult . vote ) { case "for" : stats . votesFor . add ( clientResult . client ) ; break ; case "against" : stats . votesAgainst . set ( clientResult . client , clientResult . error ) ; break ; } if ( stats . votesFor . size === stats . quorumSize ) { resolve ( { vote : "for" , stats : statsPromise , } ) ; } if ( stats . votesAgainst . size === stats . quorumSize ) { resolve ( { vote : "against" , stats : statsPromise , } ) ; } if ( stats . votesFor . size + stats . votesAgainst . size === stats . membershipSize ) { done ( ) ; } } ; const onResultReject = ( error ) : void => { throw error ; } ; for ( const result of clientResults ) { result . then ( onResultResolve , onResultReject ) ; } } ) ; } private async _attemptOperationOnClient ( client , script , keys , args ) { try { let result ; try { const shaResult = ( await client . evalsha ( script . hash , keys . length , [ ... keys , ... args , ] ) ) as unknown ; if ( typeof shaResult !== "number" ) { throw new Error ( `Unexpected result of type ${ typeof shaResult } returned from redis.` ) ; } result = shaResult ; } catch ( error ) { if ( ! ( error instanceof Error ) || ! error . message . startsWith ( "NOSCRIPT" ) ) { throw error ; } const rawResult = ( await client . eval ( script . value , keys . length , [ ... keys , ... args , ] ) ) as unknown ; if ( typeof rawResult !== "number" ) { throw new Error ( `Unexpected result of type ${ typeof rawResult } returned from redis.` ) ; } result = rawResult ; } if ( result !== keys . length ) { throw new ResourceLockedError ( `The operation was applied to: ${ result } of the ${ keys . length } requested resources.` ) ; } return { vote : "for" , client , value : result , } ; } catch ( error ) { if ( ! ( error instanceof Error ) ) { throw new Error ( `Unexpected type ${ typeof error } thrown with value: ${ error }` ) ; } this . emit ( "error" , error ) ; return { vote : "against" , client , error , } ; } } example public async using < T > ( resources , duration , settings , routine ? ) ; public async using < T > ( resources , duration , routine ) ; public async using < T > ( resources , duration , settingsOrRoutine , optionalRoutine ? ) { const settings = settingsOrRoutine && typeof settingsOrRoutine !== "function" ? { ... this . settings , ... settingsOrRoutine , } : this . settings ; const routine = optionalRoutine ?? settingsOrRoutine ; if ( typeof routine !== "function" ) { throw new Error ( "INVARIANT: routine is not a function." ) ; } if ( settings . automaticExtensionThreshold > duration - 100 ) { throw new Error ( "A lock `duration` must be at least 100ms greater than the `automaticExtensionThreshold` setting." ) ; } const controller = typeof AbortController === "undefined" ? new PolyfillAbortController ( ) : new AbortController ( ) ; const signal = controller . signal as RedlockAbortSignal ; function queue ( ) { timeout = setTimeout ( ( ) => ( extension = extend ( ) ) , lock . expiration - Date . now ( ) - settings . automaticExtensionThreshold ) ; } async function extend ( ) { timeout = undefined ; try { lock = await lock . extend ( duration ) ; queue ( ) ; } catch ( error ) { if ( ! ( error instanceof Error ) ) { throw new Error ( `Unexpected thrown ${ typeof error }: ${ error }.` ) ; } if ( lock . expiration > Date . now ( ) ) { return ( extension = extend ( ) ) ; } signal . error = error instanceof Error ? error : new Error ( `${ error }` ) ; controller . abort ( ) ; } } let timeout ; let extension ; let lock = await this . acquire ( resources , duration , settings ) ; queue ( ) ; try { return await routine ( signal ) ; } finally { if ( timeout ) { clearTimeout ( timeout ) ; timeout = undefined ; } if ( extension ) { await extension . catch ( ( ) => { } ) ; } await lock . release ( ) ; } } }