{"version":3,"sources":["common/mutexHelper.ts"],"names":[],"mappings":"AAAA,wBAAgB,KAAK,CAAC,EAAE,EAAE,MAAM,GAAG,OAAO,CAAC,IAAI,CAAC,CAU/C;AAED;;;;;;;;GAQG;AACH,wBAAgB,MAAM,CAAC,CAAC,EAAE,MAAM,EAAE,MAAM,OAAO,CAAC,CAAC,CAAC,GAAG,MAAM,OAAO,CAAC,CAAC,CAAC,CASpE;AAED,MAAM,WAAW,cAAc;IAC7B,IAAI,IAAI;QAAE,IAAI,EAAE,OAAO,CAAC;QAAC,KAAK,EAAE,MAAM,CAAA;KAAE,CAAC;IACzC,CAAC,MAAM,CAAC,QAAQ,CAAC,IAAI,cAAc,CAAC;CACrC;AAED;;;;;;;GAOG;AACH,wBAAgB,eAAe,CAAC,YAAY,KAAA,EAAE,WAAW,KAAA,GAAG,cAAc,CA8CzE;AAwBD;;;;;;;;;;;GAWG;AACH,wBAAgB,OAAO,CAAC,CAAC,EACvB,MAAM,EAAE,CAAC,GAAG,IAAI,OAAA,KAAK,OAAO,CAAC,CAAC,CAAC,EAC/B,UAAU,SAAK,EACf,SAAS,SAAM,GACd,CAAC,GAAG,IAAI,OAAA,KAAK,OAAO,CAAC,CAAC,CAAC,CAMzB;AAED;;;;;;GAMG;AACH,wBAAgB,IAAI,CAAC,CAAC,EAAE,UAAU,EAAE,MAAM,CAAC,EAAE,WAAW,GAAE,CAAC,CAAC,KAAA,KAAK,CAAY,GAAG,MAAM,CAAC,CAStF;AAED;;;GAGG;AACH,qBAAa,OAAO,CAAC,CAAC;IACpB,OAAO,CAAC,QAAQ,CAA4B;IAE5C,OAAO,CAAC,QAAQ,CAA4B;IAE5C,OAAO,CAAC,OAAO,CAAoB;IAEnC,OAAO,CAAC,YAAY,CAAgB;;IAWpC;;;;;;;SAOK;IACE,MAAM,CAAC,KAAK,EAAE,CAAC,GAAG,OAAO,CAAC,CAAC,CAAC,GAAG,OAAO;IAS7C;;;;;;;OAOG;IACI,MAAM,CAAC,GAAG,EAAE,GAAG,GAAG,OAAO;IAUhC;;SAEK;IACL,IAAI,KAAK,IAAI,MAAM,CAA8B;IAE1C,IAAI,IAAI,OAAO,CAAC,CAAC,CAAC;CAG1B;AAED;;;;GAIG;AAEH,qBAAa,KAAK;IAChB,IAAI,WAAW,WAAkC;IAEjD,IAAI,cAAc,WAAqC;IAEvD,OAAO,CAAC,iBAAiB,CAAS;IAElC,OAAO,CAAC,aAAa,CAAS;IAE9B,OAAO,CAAC,aAAa,CAAyC;IAE9D,OAAO,CAAC,eAAe,CAAK;IAE5B,OAAO,CAAC,cAAc,CAAS;IAE/B,OAAO,CAAC,UAAU,CAAK;IAEvB,OAAO,CAAC,SAAS,CAAuB;IAExC;;;;;;;SAOK;gBACO,cAAc,SAAI,EAAE,aAAa,SAAK,EAAE,WAAW,SAAK;IAMpE;;;SAGK;IACE,KAAK,CAAC,CAAC,EAAE,MAAM,EAAE,MAAM,OAAO,CAAC,CAAC,CAAC,GAAG,OAAO,CAAC,CAAC,CAAC;IAarD;;;;;;;;;SASK;IACE,QAAQ,CAAC,CAAC,EAAE,MAAM,EAAE,CAAC,GAAG,IAAI,OAAA,KAAK,OAAO,CAAC,CAAC,CAAC,GAAG,CAAC,GAAG,IAAI,OAAA,KAAK,OAAO,CAAC,CAAC,CAAC;IAM5E,OAAO,CAAC,OAAO;IA+Bf,OAAO,CAAC,IAAI;CAMb;AAED;;;;;;;;;GASG;AACH,wBAAgB,IAAI,CAAC,CAAC,EAAE,CAAC,EACvB,IAAI,EAAE,CAAC,EAAE,EACT,MAAM,EAAE,CAAC,CAAC,KAAA,KAAK,OAAO,CAAC,CAAC,CAAC,EACzB,SAAS,SAAK,EACd,MAAM,GAAE,CAAC,EAAO,GACf,OAAO,CAAC,CAAC,EAAE,CAAC,CAcd;AAED;;;;;;KAMK;AACL,wBAAgB,QAAQ,CAAC,CAAC,EAAE,KAAK,EAAE,CAAC,EAAE,MAAM,UAAQ,GAAG,CAAC,CAyBvD","file":"../../common/mutexHelper.d.ts","sourcesContent":["export function sleep(ms: number): Promise<void> {\n  return new Promise(\n    (resolve) => {\n      if (ms) {\n        setTimeout(() => resolve(), ms);\n      } else {\n        resolve();\n      }\n    },\n  );\n}\n\n/**\n * squish (debounce) the async function lambda so that\n * when a call to lambda is in flight subsequent\n * calls to lambda will short-circuit to return the\n * Promise from the already running call.\n *\n * @param lambda\n * @return squished lambda\n */\nexport function squish<T>(lambda: () => Promise<T>): () => Promise<T> {\n  let inFlight: Promise<T> = null;\n  return () => {\n    if (!inFlight) {\n      inFlight = lambda();\n      inFlight.finally(() => { inFlight = null; });\n    }\n    return inFlight;\n  };\n}\n\nexport interface NumberIterator {\n  next(): { done: boolean, value: number };\n  [Symbol.iterator](): NumberIterator;\n}\n\n/**\n * Return an iterator with the number of ms to\n * backoff before each retry\n *\n * @param maxRetries silently clamped to minimum 1, max 10\n * @param backoffMs silently clamped to minimum 100ms, max 10000ms\n * @return iterable iterator with backoff values starting from zero\n */\nexport function backoffIterator(maxRetriesIn, backoffMsIn): NumberIterator {\n  const maxRetries = (() => {\n    let valid = maxRetriesIn;\n    if (valid < 1) {\n      valid = 1;\n    }\n    if (valid > 10) {\n      valid = 10;\n    }\n    return valid;\n  })();\n  const backoffMs = (() => {\n    let valid = backoffMsIn;\n    if (valid < 100) {\n      valid = 100;\n    }\n    if (valid > 10000) {\n      valid = 10000;\n    }\n    return valid;\n  })();\n  const step = (() => {\n    let jitter = Math.floor((Math.random() * backoffMs) / 10);\n    if (Math.random() < 0.5) {\n      jitter = 0 - jitter;\n    }\n    return backoffMs + jitter;\n  })();\n\n  let count = 0;\n  let lastResult = 0;\n\n  return {\n    next() {\n      const result = {\n        done: count > maxRetries,\n        value: lastResult + lastResult,\n      };\n      if (!result.done) {\n        lastResult = result.value || step / 2;\n        count += 1;\n      }\n      return result;\n    },\n    [Symbol.iterator]() { return this; },\n  };\n}\n\n/**\n * backoff helper\n *\n * @param lambda function to backoff and retry as necessary\n * @param it iterator tracks retries\n * @param args arguments to pass through to lambda\n */\nfunction backoffProxy<T>(\n  lambda: (...inputs) => Promise<T>,\n  it: NumberIterator,\n  ...args: any[]\n): Promise<T> {\n  const next = it.next();\n  const action = sleep(next.value).then(() => lambda.apply(this, args));\n  if (next.done) {\n    return action;\n  }\n  return action.catch(\n    () => backoffProxy(lambda, it, args),\n  );\n}\n\n/**\n * Return a lambda that retries up to\n * maxRetries times with jitter exponential backoff\n * of backoffMs, 2*backoffMs, ... 2^n*backoffMs\n * capped at a 10000 ms backoff with a jitter of backoffMs/10.\n *\n * Ex: const retryFetch = backoff(function(url, options) { return fetch(url,options); });\n *\n * @param lambda\n * @param maxRetries default 3, silently clamped to minimum 1, max 10\n * @param backoffMs default 500, silently clamped to minimum 100ms, max 10000ms\n */\nexport function backoff<T>(\n  lambda: (...args) => Promise<T>,\n  maxRetries = 10,\n  backoffMs = 500,\n): (...args) => Promise<T> {\n  // eslint-disable-next-line\n  return function (...args) {\n    const it = backoffIterator(maxRetries, backoffMs);\n    return backoffProxy(lambda, it, args);\n  };\n}\n\n/**\n * Proxy that invokes lambda once, and caches the result\n *\n * @param lambdaOnce to invoke just once\n * @param lambdaTwice to invoke the second+ time called - default\n *                   just returns the given cached value\n */\nexport function once<T>(lambdaOnce: () => T, lambdaTwice: (T) => T = (t) => t): () => T {\n  let hasRun = false;\n  let cache = null;\n  return () => {\n    if (hasRun) { return lambdaTwice(cache); }\n    hasRun = true;\n    cache = lambdaOnce();\n    return cache;\n  };\n}\n\n/**\n * Barrier that resolves or rejects a promise\n * when signaled or canceled\n */\nexport class Barrier<T> {\n  private resolver: (value: T) => void = null;\n\n  private rejecter: (err: any) => void = null;\n\n  private promise: Promise<T> = null;\n\n  private barrierState = 'unresolved';\n\n  constructor() {\n    this.promise = new Promise(\n      (resolve, reject) => {\n        this.resolver = resolve;\n        this.rejecter = reject;\n      },\n    );\n  }\n\n  /**\n     * Resolve the wait() promise with value\n     *\n     * @param value\n     * @return true if value propagated to waiters,\n     *      false if barrier was already signaled\n     *      or canceled\n     */\n  public signal(value: T | Promise<T>): boolean {\n    if (this.state === 'unresolved') {\n      Promise.resolve(value).then((v) => { this.resolver(v); });\n      this.barrierState = 'resolved';\n      return true;\n    }\n    return false;\n  }\n\n  /**\n   * Propagate an error to waiters\n   *\n   * @param err\n   * @return true if value propagated to waiters,\n   *      false if barrier was already signaled\n   *      or canceled\n   */\n  public cancel(err: any): boolean {\n    if (this.state === 'unresolved') {\n      // eslnit-disable-next-line\n      this.rejecter(err);\n      this.barrierState = 'rejected';\n      return true;\n    }\n    return false;\n  }\n\n  /**\n     * @return unresolved, resolved, or rejected\n     */\n  get state(): string { return this.barrierState; }\n\n  public wait(): Promise<T> {\n    return this.promise;\n  }\n}\n\n/**\n * Helper rate limiting, circuite breaking, and mutual exclusion.\n * Note that throttle and serialize can be used in conjuction with\n * backoff and squish to retry and debounce requests.\n */\n// eslint-disable-next-line\nexport class Mutex {\n  get maxQueueLen() { return this.maxQueueLenVal; }\n\n  get maxConcurrency() { return this.maxConcurrencyVal; }\n\n  private maxConcurrencyVal: number;\n\n  private maxReqsPerSec: number;\n\n  private rateWindowEnd = new Date(Date.now() + 5000).getTime();\n\n  private rateWindowCount = 0;\n\n  private maxQueueLenVal: number;\n\n  private numRunning = 0;\n\n  private waitQueue: Barrier<void>[] = [];\n\n  /**\n     * @param maxConcurrency max number of concurrently running\n     *     requests (subsequent requests are queued) - default is 4\n     * @param maxReqsPerSec max number of requests per second\n     *     before throttling kicks in - must be greater than maxConcurrency\n     * @param maxQueueLen max length of the throttle queue before a\n     *     fast fail circuit breaker kicks in\n     */\n  constructor(maxConcurrency = 4, maxReqsPerSec = 20, maxQueueLen = 20) {\n    this.maxConcurrencyVal = maxConcurrency;\n    this.maxReqsPerSec = maxReqsPerSec;\n    this.maxQueueLenVal = maxQueueLen;\n  }\n\n  /**\n     * Aquire the mutex, then invoke lambda, then release the mutex\n     * @param lambda\n     */\n  public enter<T>(lambda: () => Promise<T>): Promise<T> {\n    // push the request onto the wait queue\n    const barrier = new Barrier<void>();\n    if (this.waitQueue.length + 1 > this.maxQueueLen) {\n      return Promise.reject(new Error('mutex throttle'));\n    }\n    this.waitQueue.push(barrier);\n    this.popTheQ();\n    const result = barrier.wait().then(() => lambda());\n    result.finally(() => this.exit());\n    return result;\n  }\n\n  /**\n     * throttle creates a proxy that limits the number of concurrently running requests\n     * to Mutex.maxConcurrency and Mutex.maxReqsPerSec by\n     * queueing them up, and circuit breaks (fast fails) requests once some\n     * queue length threshold is exceeded, so\n     * an asynchronous client could overload a backend service with requests.\n     *\n     * @param lambda that should be throttled\n     * @return throttled wrapper around lambda\n     */\n  public throttle<T>(lambda: (...args) => Promise<T>): (...args) => Promise<T> {\n    return (...args) => this.enter(\n      () => lambda(args),\n    );\n  }\n\n  private popTheQ() {\n    const now = Date.now();\n    if (now > this.rateWindowEnd) {\n      // 5 second rate limit window\n      this.rateWindowEnd = now + 5000;\n      this.rateWindowCount = 0;\n    }\n    // release rate-limitted requests\n    for (\n      let count = 0;\n      count + this.numRunning < this.maxConcurrency\n                && count + this.rateWindowCount < this.maxReqsPerSec * 5\n                && this.waitQueue.length > 0;\n      count += 1\n    ) {\n      const barrier: Barrier<void> = this.waitQueue.shift();\n      if (barrier) {\n        this.numRunning += 1;\n        this.rateWindowCount += 1;\n        barrier.signal();\n      }\n    }\n    if (this.numRunning === 0 && this.waitQueue.length > 0) {\n      // rate limiting has kicked in - wake up the queue\n      // when a new rate window opens\n      sleep(this.rateWindowEnd - now + 50).then(\n        () => this.popTheQ(),\n      );\n    }\n  }\n\n  private exit() {\n    if (this.numRunning > 0) {\n      --this.numRunning;\n    }\n    this.popTheQ();\n  }\n}\n\n/**\n * Map the given asynchronous function over the given list\n * synchronously, so that at most batchSize elements of the list\n * are in process simultaneously\n *\n * @param {[I]} list\n * @param {I => T} lambda\n * @param {int} batchSize max number of elements to run in parallel - between 1 and 100, default 10\n * @param {Promise<T>} result initial list to append to - defaults to []\n */\nexport function pmap<T, R>(\n  list: T[],\n  lambda: (T) => Promise<R>,\n  batchSize = 10,\n  result: R[] = [],\n): Promise<R[]> {\n  let n = batchSize;\n  if (n < 1) {\n    n = 1;\n  }\n  if (n > 100) {\n    n = 100;\n  }\n  if (list && list.length > 0) {\n    return Promise.all(list.slice(0, n).map((it) => lambda(it))).then(\n      (batchResult) => pmap(list.slice(n), lambda, n, result.concat(batchResult)),\n    );\n  }\n  return Promise.resolve(result);\n}\n\n/**\n   * Make a deep copy of the given object, and\n   * optionally freeze.\n   *\n   * @param thing\n   * @param freeze\n   */\nexport function deepCopy<T>(thing: T, freeze = false): T {\n  let result = null;\n\n  switch (typeof thing) {\n    case 'object':\n      if (Array.isArray(thing)) {\n        result = [];\n        for (const it of thing) {\n          result.push(deepCopy(it, freeze));\n        }\n      } else {\n        result = {};\n        for (const [key, value] of Object.entries(thing)) {\n          result[key] = deepCopy(value, freeze);\n        }\n      }\n      if (freeze) {\n        Object.freeze(result);\n      }\n      break;\n    default:\n      result = thing;\n      break;\n  }\n  return result;\n}\n"]}