daenamnu/discordeno
1
0
Fork 0
mirror of https://github.com/discordeno/discordeno.git synced 2026-09-18 01:07:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(rest): Fix followups from getting stuck in queue (#3761)

Browse Source 
* Fix followups from getting stuck in queue

* remove queueIdentifier as it isn't actually needed

* Revert some changes that aren't needed
This commit is contained in:
Fleny
2024-07-20 16:53:41 -05:00
committed by GitHub
parent 235334381c
commit 0a5493196d
3 changed files with 60 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/rest/src/manager.ts
+40 -21
View File
@@ -102,8 +102,8 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
}
},
checkRateLimits(url, requestAuthorization) {
const ratelimited = rest.rateLimitedPaths.get(`${requestAuthorization}${url}`)
checkRateLimits(url, identifier) {
const ratelimited = rest.rateLimitedPaths.get(`${identifier}${url}`)
const global = rest.rateLimitedPaths.get('global')
const now = Date.now()
@@ -147,16 +147,16 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
return
}
const newAuthorization = `Bearer ${newToken}`
const newIdentifier = `Bearer ${newToken}`
// Update all the queues
for (const [key, queue] of rest.queues.entries()) {
if (!key.startsWith(`Bearer ${oldToken}`)) continue
rest.queues.delete(key)
queue.requestAuthorization = newAuthorization
queue.identifier = newIdentifier
const newKey = `${newAuthorization}${queue.url}`
const newKey = `${newIdentifier}${queue.url}`
const newQueue = rest.queues.get(newKey)
// Merge the queues
@@ -176,10 +176,10 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
for (const [key, ratelimitPath] of rest.rateLimitedPaths.entries()) {
if (!key.startsWith(`Bearer ${oldToken}`)) continue
rest.rateLimitedPaths.set(`${newAuthorization}${ratelimitPath.url}`, ratelimitPath)
rest.rateLimitedPaths.set(`${newIdentifier}${ratelimitPath.url}`, ratelimitPath)
if (ratelimitPath.bucketId) {
rest.rateLimitedPaths.set(`${newAuthorization}${ratelimitPath.bucketId}`, ratelimitPath)
rest.rateLimitedPaths.set(`${newIdentifier}${ratelimitPath.bucketId}`, ratelimitPath)
}
}
},
@@ -320,7 +320,7 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
},
/** Processes the rate limit headers and determines if it needs to be rate limited and returns the bucket id if available */
processHeaders(url, headers, requestAuthorization) {
processHeaders(url, headers, identifier) {
let rateLimited = false
// GET ALL NECESSARY HEADERS
@@ -332,7 +332,10 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
const bucketId = headers.get(RATE_LIMIT_BUCKET_HEADER) ?? undefined
const limit = headers.get(RATE_LIMIT_LIMIT_HEADER)
rest.queues.get(`${requestAuthorization}${url}`)?.handleCompletedRequest({
// If we didn't received the identifier, fallback to the bot token
identifier ??= `Bot ${rest.token}`
rest.queues.get(`${identifier}${url}`)?.handleCompletedRequest({
remaining: remaining ? Number(remaining) : undefined,
interval: retryAfter ? Number(retryAfter) * 1000 : undefined,
max: limit ? Number(limit) : undefined,
@@ -343,7 +346,7 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
rateLimited = true
// SAVE THE URL AS LIMITED, IMPORTANT FOR NEW REQUESTS BY USER WITHOUT BUCKET
rest.rateLimitedPaths.set(`${requestAuthorization}${url}`, {
rest.rateLimitedPaths.set(`${identifier}${url}`, {
url,
resetTimestamp: reset,
bucketId,
@@ -351,7 +354,7 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
// SAVE THE BUCKET AS LIMITED SINCE DIFFERENT URLS MAY SHARE A BUCKET
if (bucketId) {
rest.rateLimitedPaths.set(`${requestAuthorization}${bucketId}`, {
rest.rateLimitedPaths.set(`${identifier}${bucketId}`, {
url,
resetTimestamp: reset,
bucketId,
@@ -380,7 +383,7 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
})
if (bucketId) {
rest.rateLimitedPaths.set(requestAuthorization, {
rest.rateLimitedPaths.set(identifier, {
url: 'global',
resetTimestamp: globalReset,
bucketId,
@@ -499,18 +502,20 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
return
}
const authorization = request.requestBodyOptions?.headers?.authorization ?? `Bot ${rest.token}`
// If we the request has a token, use it
// Else fallback to prefix with the bot token
const queueIdentifier = request.requestBodyOptions?.headers?.authorization ?? `Bot ${rest.token}`
const queue = rest.queues.get(`${authorization}${url}`)
const queue = rest.queues.get(`${queueIdentifier}${url}`)
if (queue !== undefined) {
queue.makeRequest(request)
} else {
// CREATES A NEW QUEUE
const bucketQueue = new Queue(rest, { url, deleteQueueDelay: rest.deleteQueueDelay, requestAuthorization: authorization })
const bucketQueue = new Queue(rest, { url, deleteQueueDelay: rest.deleteQueueDelay, identifier: queueIdentifier })
// Save queue
rest.queues.set(`${authorization}${url}`, bucketQueue)
rest.queues.set(`${queueIdentifier}${url}`, bucketQueue)
// Add request to queue
bucketQueue.makeRequest(request)
@@ -821,7 +826,9 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
},
async deleteWebhookWithToken(webhookId, token) {
await rest.delete(rest.routes.webhooks.webhook(webhookId, token))
await rest.delete(rest.routes.webhooks.webhook(webhookId, token), {
unauthorized: true,
})
},
async editApplicationCommandPermissions(guildId, commandId, bearerToken, permissions) {
@@ -964,11 +971,15 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
return await rest.patch<DiscordMessage>(rest.routes.webhooks.message(webhookId, token, messageId, options), {
body: options,
files: options.files,
unauthorized: true,
})
},
async editWebhookWithToken(webhookId, token, body) {
return await rest.patch<DiscordWebhook>(rest.routes.webhooks.webhook(webhookId, token), { body })
return await rest.patch<DiscordWebhook>(rest.routes.webhooks.webhook(webhookId, token), {
body,
unauthorized: true,
})
},
async editWelcomeScreen(guildId, body, reason) {
@@ -980,7 +991,10 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
},
async executeWebhook(webhookId, token, options) {
return await rest.post<DiscordMessage>(rest.routes.webhooks.webhook(webhookId, token, options), { body: options })
return await rest.post<DiscordMessage>(rest.routes.webhooks.webhook(webhookId, token, options), {
body: options,
unauthorized: true,
})
},
async followAnnouncement(sourceChannelId, targetChannelId, reason) {
@@ -1056,6 +1070,7 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
'content-type': 'application/x-www-form-urlencoded',
authorization: `Basic ${basicCredentials.toString('base64')}`,
},
runThroughQueue: false,
unauthorized: true,
}
@@ -1336,11 +1351,15 @@ export function createRestManager(options: CreateRestManagerOptions): RestManage
},
async getWebhookMessage(webhookId, token, messageId, options) {
return await rest.get<DiscordMessage>(rest.routes.webhooks.message(webhookId, token, messageId, options))
return await rest.get<DiscordMessage>(rest.routes.webhooks.message(webhookId, token, messageId, options), {
unauthorized: true,
})
},
async getWebhookWithToken(webhookId, token) {
return await rest.get<DiscordWebhook>(rest.routes.webhooks.webhook(webhookId, token))
return await rest.get<DiscordWebhook>(rest.routes.webhooks.webhook(webhookId, token), {
unauthorized: true,
})
},
async getWelcomeScreen(guildId) {
packages/rest/src/queue.ts
+18 -20
View File
@@ -31,17 +31,17 @@ export class Queue {
/** The timeout for the deletion of this queue */
deleteQueueTimeout?: NodeJS.Timeout
/**
* The authorization being used for the requests in this queue
* The identifier for this request, may be the request authorization or fallback to the bot auth
*
* @remarks
* This is also used to get the key this queue is stored as in the queue mapping of the rest manager
* This is used to get the identify this queue from the queue mapping of the rest manager
*/
requestAuthorization: string
identifier: string
constructor(rest: RestManager, options: QueueOptions) {
this.rest = rest
this.url = options.url
this.requestAuthorization = options.requestAuthorization
this.identifier = options.identifier
if (options.interval) this.interval = options.interval
if (options.max) this.max = options.max
@@ -77,7 +77,7 @@ export class Queue {
this.processing = true
while (this.waiting.length > 0) {
this.rest.logger.debug(`[Queue] ${this.getQueueType()} ${this.url} process waiting while loop ran.`)
this.rest.logger.debug(`[Queue] ${this.queueType} ${this.url} process waiting while loop ran.`)
if (this.isRequestAllowed()) {
// Resolve the next item in the queue
this.waiting.shift()?.()
@@ -99,7 +99,7 @@ export class Queue {
this.processingPending = true
while (this.pending.length > 0) {
this.rest.logger.debug(`Queue ${this.getQueueType()} ${this.url} process pending while loop ran with ${this.pending.length}.`)
this.rest.logger.debug(`Queue ${this.queueType} ${this.url} process pending while loop ran with ${this.pending.length}.`)
if (!this.firstRequest && !this.isRequestAllowed()) {
const now = Date.now()
const future = this.frozenAt + this.interval
@@ -112,11 +112,11 @@ export class Queue {
const basicURL = this.rest.simplifyUrl(request.route, request.method)
// If this url is still rate limited, try again
const urlResetIn = this.rest.checkRateLimits(basicURL, this.requestAuthorization)
const urlResetIn = this.rest.checkRateLimits(basicURL, this.identifier)
if (urlResetIn) await delay(urlResetIn)
// IF A BUCKET EXISTS, CHECK THE BUCKET'S RATE LIMITS
const bucketResetIn = request.bucketId ? this.rest.checkRateLimits(request.bucketId, this.requestAuthorization) : false
const bucketResetIn = request.bucketId ? this.rest.checkRateLimits(request.bucketId, this.identifier) : false
if (bucketResetIn) await delay(bucketResetIn)
this.firstRequest = false
@@ -134,8 +134,6 @@ export class Queue {
// Check if this request is able to be made globally
await this.rest.invalidBucket.waitUntilRequestAvailable()
if (request.requestBodyOptions?.headers?.authorization) request.requestBodyOptions.headers.authorization = this.requestAuthorization
await this.rest
.sendRequest(request)
// Should be handled in sendRequest, this catch just prevents bots from dying
@@ -143,7 +141,7 @@ export class Queue {
}
}
this.rest.logger.debug(`Queue ${this.getQueueType()} ${this.url} process pending while loop exited with ${this.pending.length}.`)
this.rest.logger.debug(`Queue ${this.queueType} ${this.url} process pending while loop exited with ${this.pending.length}.`)
// Mark as false so next pending request can be triggered by new loop.
this.processingPending = false
@@ -182,26 +180,26 @@ export class Queue {
return
}
this.rest.logger.debug(`[Queue] ${this.getQueueType()} ${this.url}. Delaying delete for ${this.deleteQueueDelay}ms`)
this.rest.logger.debug(`[Queue] ${this.queueType} ${this.url}. Delaying delete for ${this.deleteQueueDelay}ms`)
// Delete in a minute giving a bit of time to allow new requests that may reuse this queue
clearTimeout(this.deleteQueueTimeout)
this.deleteQueueTimeout = setTimeout(() => {
if (!this.isQueueClearable()) {
this.rest.logger.debug(`[Queue] ${this.getQueueType()} ${this.url}. is not clearable. Restarting processing of queue.`)
this.rest.logger.debug(`[Queue] ${this.queueType} ${this.url}. is not clearable. Restarting processing of queue.`)
this.processPending()
return
}
this.rest.logger.debug(`[Queue] ${this.getQueueType()} ${this.url}. Deleting`)
this.rest.logger.debug(`[Queue] ${this.queueType} ${this.url}. Deleting`)
if (this.timeoutId) clearTimeout(this.timeoutId)
// No requests have been requested for this queue so we nuke this queue
this.rest.queues.delete(`${this.requestAuthorization}${this.url}`)
this.rest.queues.delete(`${this.identifier}${this.url}`)
this.rest.logger.debug(
`[Queue] ${this.getQueueType()} ${this.url}. Deleted! Remaining: (${this.rest.queues.size})`,
[...this.rest.queues.values()].map((queue) => `${queue.getQueueType()}${queue.url}`),
`[Queue] ${this.queueType} ${this.url}. Deleted! Remaining: (${this.rest.queues.size})`,
[...this.rest.queues.values()].map((queue) => `${queue.queueType}${queue.url}`),
)
}, this.deleteQueueDelay)
}
@@ -217,8 +215,8 @@ export class Queue {
return true
}
getQueueType(): string {
return this.requestAuthorization.split(' ')[0]
get queueType(): string {
return this.identifier.slice(0, this.identifier.indexOf(' '))
}
}
@@ -236,5 +234,5 @@ export interface QueueOptions {
/** The time in milliseconds to wait before deleting this queue if it is empty. Defaults to 60000(one minute). */
deleteQueueDelay?: number
/** The base key that identifies this queue in the rest manager */
requestAuthorization: string
identifier: string
}
packages/rest/src/types.ts
+2 -2
View File
@@ -235,7 +235,7 @@ export interface RestManager {
/** Whether or not the rest manager should keep objects in raw snake case from discord. */
preferSnakeCase: (enabled: boolean) => RestManager
/** Check the rate limits for a url or a bucket. */
checkRateLimits: (url: string, requestAuthorization: string) => number | false
checkRateLimits: (url: string, identifier: string) => number | false
/* Update the queues and ratelimit information to adapt to the new token */
updateTokenQueues: (oldToken: string, newToken: string) => Promise<void>
/** Reshapes and modifies the obj as needed to make it ready for discords api. */
@@ -245,7 +245,7 @@ export interface RestManager {
/** This will create a infinite loop running in 1 seconds using tail recursion to keep rate limits clean. When a rate limit resets, this will remove it so the queue can proceed. */
processRateLimitedPaths: () => void
/** Processes the rate limit headers and determines if it needs to be rate limited and returns the bucket id if available */
processHeaders: (url: string, headers: Headers, requestAuthorization: string) => string | undefined
processHeaders: (url: string, headers: Headers, identifier: string) => string | undefined
/** Sends a request to the api. */
sendRequest: (options: SendRequestOptions) => Promise<void>
/** Split a url to separate rate limit buckets based on major/minor parameters. */