fix: some fixes to bucket

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>) => void> = []
+  /** The queue of requests to acquire an available request. Mapped by <shardId, resolve()> */
+  queue: Map<number, (value: void | PromiseLike<void>) => 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<void> {
+    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()
     })