import { formatWithOptions } from "util" ; import test , { ExecutionContext } from "ava" ; import Redis , { Redis as Client , Cluster } from "ioredis" ; import Redlock , { ExecutionError , ResourceLockedError } from "./index.js" ; async function fail ( t , error ) { if ( ! ( error instanceof ExecutionError ) ) { throw error ; } t . fail ( `${ error . message }
---
${ ( await Promise . all ( error . attempts ) ) . map ( ( s , i ) => `ATTEMPT ${ i }: ${ formatWithOptions ( { colors : true } , { membershipSize : s . membershipSize , quorumSize : s . quorumSize , votesForSize : s . votesFor . size , votesAgainstSize : s . votesAgainst . size , votesAgainstError : s . votesAgainst . values ( ) , } ) }` ) . join ( "\n\n" ) }
` ) ; } async function waitForCluster ( redis ) { async function checkIsReady ( ) { return ( ( ( await redis . cluster ( "info" ) ) as string ) . match ( /^cluster_state:(.+)$/m ) ?. [ 1 ] === "ok" ) ; } let isReady = await checkIsReady ( ) ; while ( ! isReady ) { console . log ( "Waiting for cluster to be ready..." ) ; await new Promise ( ( resolve ) => setTimeout ( resolve , 1000 ) ) ; isReady = await checkIsReady ( ) ; } async function checkIsWritable ( ) { try { return ( ( await redis . set ( "isWritable" , "true" ) ) as string ) === "OK" ; } catch ( error ) { console . error ( `Cluster unable to receive writes: ${ error }` ) ; return false ; } } let isWritable = await checkIsWritable ( ) ; while ( ! isWritable ) { console . log ( "Waiting for cluster to be writable..." ) ; await new Promise ( ( resolve ) => setTimeout ( resolve , 1000 ) ) ; isWritable = await checkIsWritable ( ) ; } } function run ( namespace , redis ) { test . before ( async ( ) => { await ( redis instanceof Cluster && redis . isCluster ? waitForCluster ( redis ) : null ) ; } ) ; test . before ( async ( ) => { await redis . keys ( "*" ) . then ( ( keys ) => ( keys ?. length ? redis . del ( keys ) : null ) ) ; } ) ; test ( `${ namespace } - acquires, extends, and releases a single lock` , async ( t ) => { try { const redlock = new Redlock ( [ redis ] ) ; const duration = Math . floor ( Number . MAX_SAFE_INTEGER / 10 ) ; let lock = await redlock . acquire ( [ "{redlock}a" ] , duration ) ; t . is ( await redis . get ( "{redlock}a" ) , lock . value , "The lock value was incorrect." ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}a" ) ) / 200 ) , Math . floor ( duration / 200 ) , "The lock expiration was off by more than 200ms" ) ; lock = await lock . extend ( 3 * duration ) ; t . is ( await redis . get ( "{redlock}a" ) , lock . value , "The lock value was incorrect." ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}a" ) ) / 200 ) , Math . floor ( ( 3 * duration ) / 200 ) , "The lock expiration was off by more than 200ms" ) ; await lock . release ( ) ; t . is ( await redis . get ( "{redlock}a" ) , null ) ; } catch ( error ) { fail ( t , error ) ; } } ) ; test ( `${ namespace } - acquires, extends, and releases a multi-resource lock` , async ( t ) => { try { const redlock = new Redlock ( [ redis ] ) ; const duration = Math . floor ( Number . MAX_SAFE_INTEGER / 10 ) ; let lock = await redlock . acquire ( [ "{redlock}a1" , "{redlock}a2" ] , duration ) ; t . is ( await redis . get ( "{redlock}a1" ) , lock . value , "The lock value was incorrect." ) ; t . is ( await redis . get ( "{redlock}a2" ) , lock . value , "The lock value was incorrect." ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}a1" ) ) / 200 ) , Math . floor ( duration / 200 ) , "The lock expiration was off by more than 200ms" ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}a2" ) ) / 200 ) , Math . floor ( duration / 200 ) , "The lock expiration was off by more than 200ms" ) ; lock = await lock . extend ( 3 * duration ) ; t . is ( await redis . get ( "{redlock}a1" ) , lock . value , "The lock value was incorrect." ) ; t . is ( await redis . get ( "{redlock}a2" ) , lock . value , "The lock value was incorrect." ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}a1" ) ) / 200 ) , Math . floor ( ( 3 * duration ) / 200 ) , "The lock expiration was off by more than 200ms" ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}a2" ) ) / 200 ) , Math . floor ( ( 3 * duration ) / 200 ) , "The lock expiration was off by more than 200ms" ) ; await lock . release ( ) ; t . is ( await redis . get ( "{redlock}a1" ) , null ) ; t . is ( await redis . get ( "{redlock}a2" ) , null ) ; } catch ( error ) { fail ( t , error ) ; } } ) ; test ( `${ namespace } - locks fail when redis is unreachable` , async ( t ) => { try { const redis = new Redis ( { host : "127.0.0.1" , maxRetriesPerRequest : 0 , autoResendUnfulfilledCommands : false , autoResubscribe : false , retryStrategy : ( ) => null , reconnectOnError : ( ) => false , } ) ; redis . on ( "error" , ( ) => { } ) ; const redlock = new Redlock ( [ redis ] ) ; const duration = Math . floor ( Number . MAX_SAFE_INTEGER / 10 ) ; try { await redlock . acquire ( [ "{redlock}b" ] , duration ) ; throw new Error ( "This lock should not be acquired." ) ; } catch ( error ) { if ( ! ( error instanceof ExecutionError ) ) { throw error ; } t . is ( error . attempts . length , 11 , "A failed acquisition must have the configured number of retries." ) ; for ( const e of await Promise . allSettled ( error . attempts ) ) { t . is ( e . status , "fulfilled" ) ; if ( e . status === "fulfilled" ) { for ( const v of e . value ?. votesAgainst ?. values ( ) ) { t . is ( v . message , "Connection is closed." ) ; } } } } } catch ( error ) { fail ( t , error ) ; } } ) ; test ( `${ namespace } - locks automatically expire` , async ( t ) => { try { const redlock = new Redlock ( [ redis ] ) ; const duration = 200 ; const lock = await redlock . acquire ( [ "{redlock}d" ] , duration ) ; t . is ( await redis . get ( "{redlock}d" ) , lock . value , "The lock value was incorrect." ) ; await new Promise ( ( resolve ) => setTimeout ( resolve , 300 , undefined ) ) ; const lock2 = await redlock . acquire ( [ "{redlock}d" ] , duration ) ; t . is ( await redis . get ( "{redlock}d" ) , lock2 . value , "The lock value was incorrect." ) ; await lock2 . release ( ) ; t . is ( await redis . get ( "{redlock}d" ) , null ) ; } catch ( error ) { fail ( t , error ) ; } } ) ; test ( `${ namespace } - individual locks are exclusive` , async ( t ) => { try { const redlock = new Redlock ( [ redis ] ) ; const duration = Math . floor ( Number . MAX_SAFE_INTEGER / 10 ) ; const lock = await redlock . acquire ( [ "{redlock}c" ] , duration ) ; t . is ( await redis . get ( "{redlock}c" ) , lock . value , "The lock value was incorrect." ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}c" ) ) / 200 ) , Math . floor ( duration / 200 ) , "The lock expiration was off by more than 200ms" ) ; try { await redlock . acquire ( [ "{redlock}c" ] , duration ) ; throw new Error ( "This lock should not be acquired." ) ; } catch ( error ) { if ( ! ( error instanceof ExecutionError ) ) { throw error ; } t . is ( error . attempts . length , 11 , "A failed acquisition must have the configured number of retries." ) ; for ( const e of await Promise . allSettled ( error . attempts ) ) { t . is ( e . status , "fulfilled" ) ; if ( e . status === "fulfilled" ) { for ( const v of e . value ?. votesAgainst ?. values ( ) ) { t . assert ( v instanceof ResourceLockedError , "The error must be a ResourceLockedError." ) ; } } } } await lock . release ( ) ; t . is ( await redis . get ( "{redlock}c" ) , null ) ; } catch ( error ) { fail ( t , error ) ; } } ) ; test ( `${ namespace } - overlapping multi-locks are exclusive` , async ( t ) => { try { const redlock = new Redlock ( [ redis ] ) ; const duration = Math . floor ( Number . MAX_SAFE_INTEGER / 10 ) ; const lock = await redlock . acquire ( [ "{redlock}c1" , "{redlock}c2" ] , duration ) ; t . is ( await redis . get ( "{redlock}c1" ) , lock . value , "The lock value was incorrect." ) ; t . is ( await redis . get ( "{redlock}c2" ) , lock . value , "The lock value was incorrect." ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}c1" ) ) / 200 ) , Math . floor ( duration / 200 ) , "The lock expiration was off by more than 200ms" ) ; t . is ( Math . floor ( ( await redis . pttl ( "{redlock}c2" ) ) / 200 ) , Math . floor ( duration / 200 ) , "The lock expiration was off by more than 200ms" ) ; try { await redlock . acquire ( [ "{redlock}c2" , "{redlock}c3" ] , duration ) ; throw new Error ( "This lock should not be acquired." ) ; } catch ( error ) { if ( ! ( error instanceof ExecutionError ) ) { throw error ; } t . is ( await redis . get ( "{redlock}c1" ) , lock . value , "The original lock value must not be changed." ) ; t . is ( await redis . get ( "{redlock}c2" ) , lock . value , "The original lock value must not be changed." ) ; t . is ( await redis . get ( "{redlock}c3" ) , null , "The new resource must remain unlocked." ) ; t . is ( error . attempts . length , 11 , "A failed acquisition must have the configured number of retries." ) ; for ( const e of await Promise . allSettled ( error . attempts ) ) { t . is ( e . status , "fulfilled" ) ; if ( e . status === "fulfilled" ) { for ( const v of e . value ?. votesAgainst ?. values ( ) ) { t . assert ( v instanceof ResourceLockedError , "The error must be a ResourceLockedError." ) ; } } } } await lock . release ( ) ; t . is ( await redis . get ( "{redlock}c1" ) , null ) ; t . is ( await redis . get ( "{redlock}c2" ) , null ) ; t . is ( await redis . get ( "{redlock}c3" ) , null ) ; } catch ( error ) { fail ( t , error ) ; } } ) ; test ( `${ namespace } - the \`using\` helper acquires, extends, and releases locks` , async ( t ) => { try { const redlock = new Redlock ( [ redis ] ) ; const duration = 500 ; const valueP = redlock . using ( [ "{redlock}x" ] , duration , { automaticExtensionThreshold : 200 , } , async ( signal ) => { const lockValue = await redis . get ( "{redlock}x" ) ; t . assert ( typeof lockValue === "string" , "The lock value was not correctly acquired." ) ; await new Promise ( ( resolve ) => setTimeout ( resolve , 700 , undefined ) ) ; t . is ( signal . aborted , false , "The signal must not be aborted." ) ; t . is ( signal . error , undefined , "The signal must not have an error." ) ; t . is ( await redis . get ( "{redlock}x" ) , lockValue , "The lock value should not have changed." ) ; return lockValue ; } ) ; await valueP ; t . is ( await redis . get ( "{redlock}x" ) , null , "The lock was not released." ) ; } catch ( error ) { fail ( t , error ) ; } } ) ; test ( `${ namespace } - the \`using\` helper is exclusive` , async ( t ) => { try { const redlock = new Redlock ( [ redis ] ) ; const duration = 500 ; let locked = false ; const [ lock1 , lock2 ] = await Promise . all ( [ await redlock . using ( [ "{redlock}y" ] , duration , { automaticExtensionThreshold : 200 , } , async ( signal ) => { t . is ( locked , false , "The resource must not already be locked." ) ; locked = true ; const lockValue = await redis . get ( "{redlock}y" ) ; t . assert ( typeof lockValue === "string" , "The lock value was not correctly acquired." ) ; await new Promise ( ( resolve ) => setTimeout ( resolve , 700 , undefined ) ) ; t . is ( signal . error , undefined , "The signal must not have an error." ) ; t . is ( signal . aborted , false , "The signal must not be aborted." ) ; t . is ( await redis . get ( "{redlock}y" ) , lockValue , "The lock value should not have changed." ) ; locked = false ; return lockValue ; } ) , await redlock . using ( [ "{redlock}y" ] , duration , { automaticExtensionThreshold : 200 , } , async ( signal ) => { t . is ( locked , false , "The resource must not already be locked." ) ; locked = true ; const lockValue = await redis . get ( "{redlock}y" ) ; t . assert ( typeof lockValue === "string" , "The lock value was not correctly acquired." ) ; await new Promise ( ( resolve ) => setTimeout ( resolve , 700 , undefined ) ) ; t . is ( signal . error , undefined , "The signal must not have an error." ) ; t . is ( signal . aborted , false , "The signal must not be aborted." ) ; t . is ( await redis . get ( "{redlock}y" ) , lockValue , "The lock value should not have changed." ) ; locked = false ; return lockValue ; } ) , ] ) ; t . not ( lock1 , lock2 , "The locks must be different." ) ; t . is ( await redis . get ( "{redlock}y" ) , null , "The lock was not released." ) ; } catch ( error ) { fail ( t , error ) ; } } ) ; } run ( "instance" , new Redis ( { host : "redis-single-instance" } ) ) ; run ( "cluster" , new Cluster ( [ { host : "redis-single-cluster-1" } ] ) ) ;