diff --git a/packages/utils/src/bucket.ts b/packages/utils/src/bucket.ts index f8f153c4b..ad9da51da 100644 --- a/packages/utils/src/bucket.ts +++ b/packages/utils/src/bucket.ts @@ -1,3 +1,4 @@ +import logger from './logger.js' import { delay } from './utils.js' export class LeakyBucket implements LeakyBucketOptions { @@ -7,8 +8,8 @@ export class LeakyBucket implements LeakyBucketOptions { /** The amount of requests that have been used up already. */ used: number = 0 - /** The queue of requests to acquire an available request. */ - queue: Array<(value: void | PromiseLike) => void> = [] + /** The queue of requests to acquire an available request. Mapped by */ + queue: Map) => void> = new Map() /** 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. */ @@ -18,7 +19,7 @@ export class LeakyBucket implements LeakyBucketOptions { constructor(options?: LeakyBucketOptions) { this.max = options?.max ?? 1 - this.refillAmount = options?.refillAmount ? options.refillAmount > this.max ? this.max : options.refillAmount : 1 + this.refillAmount = options?.refillAmount ? (options.refillAmount > this.max ? this.max : options.refillAmount) : 1 this.refillInterval = options?.refillInterval ?? 5000 } @@ -29,12 +30,17 @@ export class LeakyBucket implements LeakyBucketOptions { /** Begin processing the queue. */ async processQueue(): Promise { + logger.debug('[Gateway] Processing queue') // There is already a queue that is processing - if (this.processing) return + if (this.processing) { + logger.debug('[Gateway] Queue is already processing.') + return + } // Begin going through the queue. while (this.queue.length) { if (this.remaining) { + logger.debug(`[LeakyBucket] Processing queue. Remaining: ${this.remaining} Length: ${this.queue.length}`) // Resolves the promise allowing the paused execution of this request to resolve and continue. this.queue.shift()?.() // A request can be made @@ -42,7 +48,9 @@ export class LeakyBucket implements LeakyBucketOptions { // Create a new timeout for this request if none exists. if (!this.timeoutId) { + logger.debug(`[LeakyBucket] Creating new timeout for leaky bucket requests.`) this.timeoutId = setTimeout(() => { + logger.debug(`[LeakyBucket] Timeout for leaky bucket requests executed. Refilling bucket.`) // Lower the used amount by the refill amount this.used -= this.refillAmount // Reset the refillsAt timestamp since it just got refilled @@ -54,14 +62,19 @@ export class LeakyBucket implements LeakyBucketOptions { } // Check if a refill is scheduled, since we have used up all available requests - if (this.refillsAt) { + else if (this.refillsAt) { const now = Date.now() // If there is time left until next refill, just delay execution. if (this.refillsAt > now) { + logger.debug(`[LeakyBucket] Delaying execution of leaky bucket requests for ${this.refillsAt - now}ms`) await delay(this.refillsAt - now) + logger.debug(`[LeakyBucket] Resuming execution`) } } } + + // Loop has ended mark false so it can restart later when needed + this.processing = false; } /** Pauses the execution until the request is available to be made. */ @@ -71,6 +84,7 @@ export class LeakyBucket implements LeakyBucketOptions { if (highPriority) this.queue.unshift(resolve) // All other requests get pushed to the end. else this.queue.push(resolve) + // Each request should trigger the queue to be processesd. void this.processQueue() })