diff --git a/packages/utils/src/bucket.ts b/packages/utils/src/bucket.ts index c23a61e75..1b1e85544 100644 --- a/packages/utils/src/bucket.ts +++ b/packages/utils/src/bucket.ts @@ -12,7 +12,7 @@ export class LeakyBucket implements LeakyBucketOptions { queue: Array<(value: void | PromiseLike) => void> = [] /** Whether or not the queue is already processing. */ processing: boolean = false - /** The timeout id for the timer to reduce the used amount by the refill amount. */ + /** The timeout id for the timer to reduce the used amount by the refill amount. */ timeoutId?: NodeJS.Timeout /** The timestamp in milliseconds when the next refill is scheduled. */ refillsAt?: number @@ -37,7 +37,7 @@ export class LeakyBucket implements LeakyBucketOptions { this.refillsAt = undefined // Reset the timeoutId clearTimeout(this.timeoutId) - this.timeoutId = undefined; + this.timeoutId = undefined if (this.used > 0) { this.timeoutId = setTimeout(() => { @@ -49,9 +49,12 @@ export class LeakyBucket implements LeakyBucketOptions { /** Begin processing the queue. */ async processQueue(): Promise { - logger.debug('[Gateway] Processing queue') + logger.debug('[LeakyBucket] Processing queue') + // There is already a queue that is processing - if (this.processing) return logger.debug('[Gateway] Queue is already processing.') + if (this.processing) return logger.debug('[LeakyBucket] Queue is already processing.') + + this.processing = true // Begin going through the queue. while (this.queue.length) { @@ -83,6 +86,12 @@ export class LeakyBucket implements LeakyBucketOptions { await delay(this.refillsAt - now) logger.debug(`[LeakyBucket] Resuming execution`) } + + // If the refillsAt has passed but the timeout didn't yet execute delay the execution + else { + logger.debug(`[LeakyBucket] Delaying execution of leaky bucket requests for 1000ms`) + await delay(1000) + } } } @@ -98,7 +107,7 @@ export class LeakyBucket implements LeakyBucketOptions { // All other requests get pushed to the end. else this.queue.push(resolve) - // Each request should trigger the queue to be processesd. + // Each request should trigger the queue to be processed. void this.processQueue() }) }