feat: added acquireAttemptsLimit

Author: swarthy

README.md

@@ -28,13 +28,14 @@ yarn add redis-semaphore ioredis
 
 > See [RedisLabs: Locks with timeouts](https://redislabs.com/ebook/part-2-core-concepts/chapter-6-application-components-in-redis/6-2-distributed-locking/6-2-5-locks-with-timeouts/)
 
-##### new Mutex(redisClient, key [, { lockTimeout = 10000, acquireTimeout = 10000, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
+##### new Mutex(redisClient, key [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
 
 - `redisClient` - **required**, configured `redis` client
 - `key` - **required**, key for locking resource (final key in redis: `mutex:<key>`)
 - `options` - _optional_
  - `lockTimeout` - _optional_ ms, time after mutex will be auto released (expired)
  - `acquireTimeout` - _optional_ ms, max timeout for `.acquire()` call
+ - `acquireAttemptsLimit` - _optional_ max number of attempts to be made in `.acquire()` call
  - `retryInterval` - _optional_ ms, time between acquire attempts if resource locked
  - `refreshInterval` - _optional_ ms, auto-refresh interval; to disable auto-refresh behaviour set `0`
  - `onLockLost` - _optional_ function, called when lock loss is detected due refresh cycle; default onLockLost throws unhandled LostLockError
@@ -88,7 +89,9 @@ async function doSomething() {
 
 ```javascript
 async function doSomething() {
- const mutex = new Mutex(redisClient, 'lockingResource', { acquireTimeout: 0 })
+ const mutex = new Mutex(redisClient, 'lockingResource', {
+ acquireAttemptsLimit: 1
+ })
  const lockAcquired = await mutex.tryAcquire()
  if (!lockAcquired) {
  return
@@ -116,7 +119,7 @@ In edge cases (node time difference is greater than `lockTimeout`) both algorith
 
 Most reliable way to use: `lockTimeout` is greater than possible node clock differences, `refreshInterval` is not 0 and is less enough than `lockTimeout` (by default is `lockTimeout * 0.8`)
 
-##### new Semaphore(redisClient, key, maxCount [, { lockTimeout = 10000, acquireTimeout = 10000, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
+##### new Semaphore(redisClient, key, maxCount [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
 
 - `redisClient` - **required**, configured `redis` client
 - `key` - **required**, key for locking resource (final key in redis: `semaphore:<key>`)
@@ -152,7 +155,7 @@ Same as `Semaphore` with one difference - MultiSemaphore will try to acquire mul
 
 `MultiSemaphore` and `Semaphore` shares same key namespace and can be used together (see test/src/RedisMultiSemaphore.test.ts).
 
-##### new MultiSemaphore(redisClient, key, maxCount, permits [, { lockTimeout = 10000, acquireTimeout = 10000, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
+##### new MultiSemaphore(redisClient, key, maxCount, permits [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
 
 - `redisClient` - **required**, configured `redis` client
 - `key` - **required**, key for locking resource (final key in redis: `semaphore:<key>`)
@@ -190,7 +193,7 @@ Distributed `Mutex` version
 
 > See [The Redlock algorithm](https://redis.io/topics/distlock#the-redlock-algorithm)
 
-##### new RedlockMutex(redisClients, key [, { lockTimeout = 10000, acquireTimeout = 10000, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
+##### new RedlockMutex(redisClients, key [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
 
 - `redisClients` - **required**, array of configured `redis` client connected to independent nodes
 - `key` - **required**, key for locking resource (final key in redis: `mutex:<key>`)
@@ -229,7 +232,7 @@ Distributed `Semaphore` version
 
 > See [The Redlock algorithm](https://redis.io/topics/distlock#the-redlock-algorithm)
 
-##### new RedlockSemaphore(redisClients, key, maxCount [, { lockTimeout = 10000, acquireTimeout = 10000, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
+##### new RedlockSemaphore(redisClients, key, maxCount [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
 
 - `redisClients` - **required**, array of configured `redis` client connected to independent nodes
 - `key` - **required**, key for locking resource (final key in redis: `semaphore:<key>`)
@@ -269,7 +272,7 @@ Distributed `MultiSemaphore` version
 
 > See [The Redlock algorithm](https://redis.io/topics/distlock#the-redlock-algorithm)
 
-##### new RedlockMultiSemaphore(redisClients, key, maxCount, permits [, { lockTimeout = 10000, acquireTimeout = 10000, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
+##### new RedlockMultiSemaphore(redisClients, key, maxCount, permits [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])
 
 - `redisClients` - **required**, array of configured `redis` client connected to independent nodes
 - `key` - **required**, key for locking resource (final key in redis: `semaphore:<key>`)

src/Lock.ts

@@ -4,19 +4,12 @@ import { v4 as uuid4 } from 'uuid'
 import LostLockError from './errors/LostLockError'
 import TimeoutError from './errors/TimeoutError'
 import { defaultOnLockLost, defaultTimeoutOptions } from './misc'
-import { LockLostCallback, LockOptions } from './types'
+import { AcquireOptions, LockLostCallback, LockOptions } from './types'
 
 const REFRESH_INTERVAL_COEF = 0.8
 
 const debug = createDebug('redis-semaphore:instance')
 
-interface AcquireOptions {
- identifier: string
- lockTimeout: number
- acquireTimeout: number
- retryInterval: number
-}
-
 export abstract class Lock {
  protected abstract _kind: string
  protected abstract _key: string
@@ -35,6 +28,7 @@ export abstract class Lock {
  constructor({
  lockTimeout = defaultTimeoutOptions.lockTimeout,
  acquireTimeout = defaultTimeoutOptions.acquireTimeout,
+ acquireAttemptsLimit = defaultTimeoutOptions.acquireAttemptsLimit,
  retryInterval = defaultTimeoutOptions.retryInterval,
  refreshInterval = Math.round(lockTimeout * REFRESH_INTERVAL_COEF),
  onLockLost = defaultOnLockLost
@@ -43,6 +37,7 @@ export abstract class Lock {
  this._acquireOptions = {
  lockTimeout,
  acquireTimeout,
+ acquireAttemptsLimit,
  retryInterval,
  identifier: this._identifier
  }

src/misc.ts

@@ -3,6 +3,7 @@ import LostLockError from '../src/errors/LostLockError'
 export const defaultTimeoutOptions = {
  lockTimeout: 10000,
  acquireTimeout: 10000,
+ acquireAttemptsLimit: Number.POSITIVE_INFINITY,
  retryInterval: 10
 }
 

src/multiSemaphore/acquire/index.ts

@@ -10,6 +10,7 @@ export interface Options {
  identifier: string
  lockTimeout: number
  acquireTimeout: number
+ acquireAttemptsLimit: number
  retryInterval: number
 }
 
@@ -20,11 +21,18 @@ export async function acquireSemaphore(
  permits: number,
  options: Options
 ) {
- const { identifier, lockTimeout, acquireTimeout, retryInterval } = options
+ const {
+ identifier,
+ lockTimeout,
+ acquireTimeout,
+ acquireAttemptsLimit,
+ retryInterval
+ } = options
+ let attempt = 0
  const end = Date.now() + acquireTimeout
  let now
- while ((now = Date.now()) < end) {
- debug(key, identifier, limit, lockTimeout)
+ while ((now = Date.now()) < end && ++attempt <= acquireAttemptsLimit) {
+ debug(key, identifier, limit, lockTimeout, 'attempt', attempt)
  const result = await acquireLua(client, [
  key,
  limit,
@@ -41,5 +49,6 @@ export async function acquireSemaphore(
  await delay(retryInterval)
  }
  }
+ debug(key, identifier, limit, lockTimeout, 'timeout or reach limit')
  return false
 }

src/mutex/acquire.ts

@@ -9,6 +9,7 @@ export interface Options {
  identifier: string
  lockTimeout: number
  acquireTimeout: number
+ acquireAttemptsLimit: number
  retryInterval: number
 }
 
@@ -17,10 +18,17 @@ export async function acquireMutex(
  key: string,
  options: Options
 ) {
- const { identifier, lockTimeout, acquireTimeout, retryInterval } = options
+ const {
+ identifier,
+ lockTimeout,
+ acquireTimeout,
+ acquireAttemptsLimit,
+ retryInterval
+ } = options
+ let attempt = 0
  const end = Date.now() + acquireTimeout
- while (Date.now() < end) {
- debug(key, identifier, 'attempt')
+ while (Date.now() < end && ++attempt <= acquireAttemptsLimit) {
+ debug(key, identifier, 'attempt', attempt)
  const result = await client.set(key, identifier, 'PX', lockTimeout, 'NX')
  debug('result', typeof result, result)
  if (result === 'OK') {
@@ -30,6 +38,6 @@ export async function acquireMutex(
  await delay(retryInterval)
  }
  }
- debug(key, identifier, 'timeout')
+ debug(key, identifier, 'timeout or reach limit')
  return false
 }

src/redlockMultiSemaphore/acquire.ts

@@ -11,6 +11,7 @@ export interface Options {
  identifier: string
  lockTimeout: number
  acquireTimeout: number
+ acquireAttemptsLimit: number
  retryInterval: number
 }
 
@@ -21,12 +22,19 @@ export async function acquireRedlockMultiSemaphore(
  permits: number,
  options: Options
 ) {
- const { identifier, lockTimeout, acquireTimeout, retryInterval } = options
+ const {
+ identifier,
+ lockTimeout,
+ acquireTimeout,
+ acquireAttemptsLimit,
+ retryInterval
+ } = options
+ let attempt = 0
  const end = Date.now() + acquireTimeout
  const quorum = getQuorum(clients.length)
  let now: number
- while ((now = Date.now()) < end) {
- debug(key, identifier, 'attempt')
+ while ((now = Date.now()) < end && ++attempt <= acquireAttemptsLimit) {
+ debug(key, identifier, 'attempt', attempt)
  const promises = clients.map(client =>
  acquireLua(client, [key, limit, permits, identifier, lockTimeout, now])
  .then(result => +result)
@@ -44,6 +52,6 @@ export async function acquireRedlockMultiSemaphore(
  await delay(retryInterval)
  }
  }
- debug(key, identifier, 'timeout')
+ debug(key, identifier, 'timeout or reach limit')
  return false
 }

src/redlockMutex/acquire.ts

@@ -11,6 +11,7 @@ export interface Options {
  identifier: string
  lockTimeout: number
  acquireTimeout: number
+ acquireAttemptsLimit: number
  retryInterval: number
 }
 
@@ -19,11 +20,18 @@ export async function acquireRedlockMutex(
  key: string,
  options: Options
 ) {
- const { identifier, lockTimeout, acquireTimeout, retryInterval } = options
+ const {
+ identifier,
+ lockTimeout,
+ acquireTimeout,
+ acquireAttemptsLimit,
+ retryInterval
+ } = options
+ let attempt = 0
  const end = Date.now() + acquireTimeout
  const quorum = getQuorum(clients.length)
- while (Date.now() < end) {
- debug(key, identifier, 'attempt')
+ while (Date.now() < end && ++attempt <= acquireAttemptsLimit) {
+ debug(key, identifier, 'attempt', attempt)
  const promises = clients.map(client =>
  client
  .set(key, identifier, 'PX', lockTimeout, 'NX')
@@ -42,6 +50,6 @@ export async function acquireRedlockMutex(
  await delay(retryInterval)
  }
  }
- debug(key, identifier, 'timeout')
+ debug(key, identifier, 'timeout or reach limit')
  return false
 }

src/redlockSemaphore/acquire.ts

@@ -11,6 +11,7 @@ export interface Options {
  identifier: string
  lockTimeout: number
  acquireTimeout: number
+ acquireAttemptsLimit: number
  retryInterval: number
 }
 
@@ -20,12 +21,19 @@ export async function acquireRedlockSemaphore(
  limit: number,
  options: Options
 ) {
- const { identifier, lockTimeout, acquireTimeout, retryInterval } = options
+ const {
+ identifier,
+ lockTimeout,
+ acquireTimeout,
+ acquireAttemptsLimit,
+ retryInterval
+ } = options
+ let attempt = 0
  const end = Date.now() + acquireTimeout
  const quorum = getQuorum(clients.length)
  let now: number
- while ((now = Date.now()) < end) {
- debug(key, identifier, 'attempt')
+ while ((now = Date.now()) < end && ++attempt <= acquireAttemptsLimit) {
+ debug(key, identifier, 'attempt', attempt)
  const promises = clients.map(client =>
  acquireLua(client, [key, limit, identifier, lockTimeout, now])
  .then(result => +result)
@@ -43,6 +51,6 @@ export async function acquireRedlockSemaphore(
  await delay(retryInterval)
  }
  }
- debug(key, identifier, 'timeout')
+ debug(key, identifier, 'timeout or reach limit')
  return false
 }

src/semaphore/acquire/index.ts

@@ -10,6 +10,7 @@ export interface Options {
  identifier: string
  lockTimeout: number
  acquireTimeout: number
+ acquireAttemptsLimit: number
  retryInterval: number
 }
 
@@ -19,11 +20,18 @@ export async function acquireSemaphore(
  limit: number,
  options: Options
 ) {
- const { identifier, lockTimeout, acquireTimeout, retryInterval } = options
+ const {
+ identifier,
+ lockTimeout,
+ acquireTimeout,
+ acquireAttemptsLimit,
+ retryInterval
+ } = options
+ let attempt = 0
  const end = Date.now() + acquireTimeout
  let now
- while ((now = Date.now()) < end) {
- debug(key, identifier, limit, lockTimeout)
+ while ((now = Date.now()) < end && ++attempt <= acquireAttemptsLimit) {
+ debug(key, identifier, limit, lockTimeout, 'attempt', attempt)
  const result = await acquireLua(client, [
  key,
  limit,
@@ -40,5 +48,6 @@ export async function acquireSemaphore(
  await delay(retryInterval)
  }
  }
+ debug(key, identifier, limit, lockTimeout, 'timeout or reach limit')
  return false
 }

src/types.ts

@@ -8,10 +8,19 @@ export interface LockLostCallback {
 export interface TimeoutOptions {
  lockTimeout?: number
  acquireTimeout?: number
+ acquireAttemptsLimit?: number
  retryInterval?: number
  refreshInterval?: number
 }
 
 export interface LockOptions extends TimeoutOptions {
  onLockLost?: LockLostCallback
 }
+
+export interface AcquireOptions {
+ identifier: string
+ lockTimeout: number
+ acquireTimeout: number
+ acquireAttemptsLimit: number
+ retryInterval: number
+}