From 3d12ce28530faf61c90867b920f4bec9ae45338e Mon Sep 17 00:00:00 2001 From: Skillz Date: Sun, 12 Mar 2023 15:39:04 -0500 Subject: [PATCH] fix: add logs and shard queue sorting --- packages/gateway/src/Shard.ts | 13 ++++++------- packages/gateway/src/manager.ts | 17 ++++++++++++----- packages/utils/src/bucket.ts | 10 +++++----- 3 files changed, 23 insertions(+), 17 deletions(-) diff --git a/packages/gateway/src/Shard.ts b/packages/gateway/src/Shard.ts index 8c30ed198..40e3e64f2 100644 --- a/packages/gateway/src/Shard.ts +++ b/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() diff --git a/packages/gateway/src/manager.ts b/packages/gateway/src/manager.ts index 8abbcacfb..1baacda68 100644 --- a/packages/gateway/src/manager.ts +++ b/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 diff --git a/packages/utils/src/bucket.ts b/packages/utils/src/bucket.ts index 9e9776d3f..05a8ba0c4 100644 --- a/packages/utils/src/bucket.ts +++ b/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 */ - queue: Array<{ shardId: number; resolve: (value: void | PromiseLike) => void }> = [] + 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. */ @@ -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 { + async acquire(highPriority?: boolean): Promise { 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()