fix: add logs and shard queue sorting

packages/gateway/src/Shard.ts

@@ -194,7 +194,7 @@ export class DiscordenoShard {
       this.resolves.set('READY', () => {
         this.events.identified?.(this)
         // Tells the manager that this shard is ready
-        this.shardIsReady();
+        this.shardIsReady()
         resolve()
       })
       // When identifying too fast,
@@ -223,10 +223,7 @@ export class DiscordenoShard {
 
     // Shard has never identified, so we cannot resume.
     if (!this.sessionId) {
-      // gateway.debug(
-      //   "GW DEBUG",
-      //   `[Error] Trying to resume a shard (id: ${shardId}) that was not first identified.`,
-      // );
+      logger.debug(`[Shard] Trying to resume a shard #${this.id} that was NOT first identified. (No session id found)`)
 
       return await this.identify()
 
@@ -270,7 +267,7 @@ export class DiscordenoShard {
     // Else bucket and token wait time just get wasted.
     await this.checkOffline(highPriority)
 
-    await this.bucket.acquire(this.id, highPriority)
+    await this.bucket.acquire(highPriority)
 
     // It's possible, that the shard went offline after a token has been acquired from the bucket.
     await this.checkOffline(highPriority)
@@ -315,6 +312,7 @@ export class DiscordenoShard {
       case GatewayCloseEventCodes.InvalidSeq:
       case GatewayCloseEventCodes.RateLimited:
       case GatewayCloseEventCodes.SessionTimedOut: {
+        logger.debug(`[Shard] Gateway connection closing requiring re-identify. Code: ${close.code}`)
         this.state = ShardState.Identifying
         this.events.disconnected?.(this)
 
@@ -414,8 +412,8 @@ export class DiscordenoShard {
         break
       }
       case GatewayOpcodes.InvalidSession: {
-        //   gateway.debug("GW INVALID_SESSION", { shardId, payload: packet });
         const resumable = packet.d as boolean
+        logger.debug(`[Shard] Received Invalid Session for Shard #${this.id} with resumeable as ${resumable.toString()}`)
 
         this.events.invalidSession?.(this, resumable)
 
@@ -556,6 +554,7 @@ export class DiscordenoShard {
         // The Shard needs to start a re-identify action accordingly.
         // Reference: https://discord.com/developers/docs/topics/gateway#heartbeating-example-gateway-heartbeat-ack
         if (!this.heart.acknowledged) {
+          logger.debug(`[Shard] Heartbeat not acknowledged for shard #${this.id}.`)
           this.close(ShardSocketCloseCodes.ZombiedConnection, 'Zombied connection, did not receive an heartbeat ACK in time.')
 
           return await this.identify()

packages/gateway/src/manager.ts

@@ -114,6 +114,13 @@ export function createGatewayManager(options: CreateGatewayManagerOptions): Gate
           bucket.workers.push({ id: workerId, queue: [shardId] })
         }
       }
+
+      for (const bucket of gateway.buckets.values()) {
+        for (const worker of bucket.workers.values()) {
+          // eslint-disable-next-line @typescript-eslint/require-array-sort-compare
+          worker.queue = worker.queue.sort((a, b) => a - b)
+        }
+      }
     },
     async spawnShards() {
       // PREPARES ALL SHARDS IN SPECIFIC BUCKETS
@@ -195,12 +202,12 @@ export function createGatewayManager(options: CreateGatewayManagerOptions): Gate
 
     async requestIdentify(shardId: number) {
       logger.debug(`[Gateway] requesting identify`)
-      const bucket = gateway.buckets.get(shardId % gateway.connection.sessionStartLimit.maxConcurrency)
-      if (!bucket) return
+      // const bucket = gateway.buckets.get(shardId % gateway.connection.sessionStartLimit.maxConcurrency)
+      // if (!bucket) return
 
-      return await new Promise((resolve) => {
-        bucket.identifyRequests.push(resolve)
-      })
+      // return await new Promise((resolve) => {
+      //   bucket.identifyRequests.push(resolve)
+      // })
     },
 
     // Helpers methods below this

packages/utils/src/bucket.ts

@@ -9,7 +9,7 @@ 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. Mapped by <shardId, resolve()> */
-  queue: Array<{ shardId: number; resolve: (value: void | PromiseLike<void>) => void }> = []
+  queue: Array<(value: void | PromiseLike<void>) => 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.  */
@@ -42,7 +42,7 @@ export class LeakyBucket implements LeakyBucketOptions {
       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()?.resolve()
+        this.queue.shift()?.()
         // A request can be made
         this.used++
 
@@ -78,12 +78,12 @@ export class LeakyBucket implements LeakyBucketOptions {
   }
 
   /** Pauses the execution until the request is available to be made. */
-  async acquire(shardId: number, highPriority?: boolean): Promise<void> {
+  async acquire(highPriority?: boolean): Promise<void> {
     return await new Promise((resolve) => {
       // High priority requests get added to the start of the queue
-      if (highPriority) this.queue.unshift({ shardId, resolve })
+      if (highPriority) this.queue.unshift(resolve)
       // All other requests get pushed to the end.
-      else this.queue.push({ shardId, resolve })
+      else this.queue.push(resolve)
 
       // Each request should trigger the queue to be processesd.
       void this.processQueue()