-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: rewrite concurrency manager (#8)
* feat: rewrite concurrency manager * fix: missing auth header * feat: implement abort identifies * feat: implement server using vanilla http * fix: POST not DELETE op in acquire * fix: Return no content as status message * fix: fix start issues * fix: fix http dont throw errors now * fix: address is a function not a prop * fix: force listen to localhost ipv4 * feat: add debug logs for concurrency server * fix: make fetch aware of http or https * fix: make https first over http * fix: query string parsing * chore: remove deps once again * refactor: use event emitter raw * fix: use NodeJS.EventEmitter for event emitter type
- Loading branch information
Showing
8 changed files
with
207 additions
and
59 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,44 +1,62 @@ | ||
| import { InternalOps, InternalOpsData } from '../Util'; | ||
| import { BaseWorker } from '../ipc/BaseWorker'; | ||
| import { Delay, Fetch } from '../Util'; | ||
|
|
||
| /** | ||
| * Internal class that is passed to @discordjs/ws to handle concurrency | ||
| */ | ||
| export class ConcurrencyClient { | ||
| private ipc: BaseWorker; | ||
| constructor(ipc: BaseWorker) { | ||
| this.ipc = ipc; | ||
| private readonly address: string; | ||
| private readonly port: number; | ||
| private readonly password: string; | ||
|
|
||
| constructor() { | ||
| this.address = process.env.INDOMITABLE_CONCURRENCY_SERVER_ADDRESS!; | ||
| this.port = Number(process.env.INDOMITABLE_CONCURRENCY_SERVER_PORT!); | ||
| this.password = process.env.INDOMITABLE_CONCURRENCY_SERVER_PASSWORD!; | ||
| } | ||
|
|
||
| /** | ||
| * Method to try and acquire a lock for identify | ||
| * Method to try and acquire a lock for identify. This could never error or else it would hang out the whole system. | ||
| * Look at (https://github.com/discordjs/discord.js/blob/f1bce54a287eaa431ceb8b1996db87cbc6290317/packages/ws/src/strategies/sharding/WorkerShardingStrategy.ts#L321) | ||
| * If it errors that isn't anything from websocket shard, this will have issues | ||
| */ | ||
| public async waitForIdentify(shardId: number, signal: AbortSignal): Promise<void> { | ||
| const content: InternalOpsData = { | ||
| op: InternalOps.REQUEST_IDENTIFY, | ||
| data: { shardId }, | ||
| internal: true | ||
| }; | ||
| const listener = () => this.abortIdentify(shardId); | ||
| const url = new URL(`http://${this.address}:${this.port}/concurrency/acquire`); | ||
| url.searchParams.append('shardId', shardId.toString()); | ||
|
|
||
| const listener = () => { | ||
| const url = new URL(`http://${this.address}:${this.port}/concurrency/cancel`); | ||
|
|
||
| url.searchParams.append('shardId', shardId.toString()); | ||
|
|
||
| Fetch(url.toString(), { | ||
| method: 'DELETE', | ||
| headers: { authorization: this.password } | ||
| }).catch(() => null); | ||
| } | ||
|
|
||
| try { | ||
| signal.addEventListener('abort', listener); | ||
| await this.ipc.send({ content, repliable: true }); | ||
|
|
||
| const response = await Fetch(url.toString(), { | ||
| method: 'POST', | ||
| headers: { authorization: this.password } | ||
| }); | ||
|
|
||
| if (response.code === 202 || response.code === 204) { | ||
| // aborted request || ok request | ||
| return; | ||
| } | ||
|
|
||
| if (response.code >= 400 && response.code <= 499) { | ||
| // something happened server didn't accept your req | ||
| await Delay(1000); | ||
| return; | ||
| } | ||
| } catch (_) { | ||
| // this should not happen but we just delay if it happens | ||
| await Delay(1000); | ||
| } finally { | ||
| signal.removeEventListener('abort', listener); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Aborts an acquire lock request | ||
| */ | ||
| private abortIdentify(shardId: number): void { | ||
| const content: InternalOpsData = { | ||
| op: InternalOps.CANCEL_IDENTIFY, | ||
| data: { shardId }, | ||
| internal: true | ||
| }; | ||
| this.ipc | ||
| .send({ content, repliable: false }) | ||
| .catch(() => null); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| import { AddressInfo } from 'node:net'; | ||
| import { ConcurrencyManager } from './ConcurrencyManager'; | ||
| import { Indomitable } from '../Indomitable'; | ||
| import { LibraryEvents } from '../Util'; | ||
| import Http from 'node:http'; | ||
| import QueryString from 'node:querystring'; | ||
|
|
||
| /** | ||
| * Server that handles identify locks | ||
| */ | ||
| export class ConcurrencyServer { | ||
| private readonly manager: Indomitable; | ||
| /** | ||
| * Http server of this instance | ||
| * @private | ||
| */ | ||
| private readonly server: Http.Server; | ||
| /** | ||
| * Concurrency manager for this server | ||
| * @private | ||
| */ | ||
| private readonly concurrency: ConcurrencyManager; | ||
| /** | ||
| * Randomly generated password to secure this server | ||
| * @private | ||
| */ | ||
| private readonly password: string; | ||
|
|
||
| constructor(manager: Indomitable, concurrency: number) { | ||
| this.manager = manager; | ||
| this.server = Http.createServer((req, res) => this.handle(req, res)); | ||
| this.concurrency = new ConcurrencyManager(concurrency); | ||
| this.password = Math.random().toString(36).slice(2, 10); | ||
| } | ||
|
|
||
| /** | ||
| * Gets the randomly generated password for this instance | ||
| */ | ||
| public get key(): string { | ||
| return this.password; | ||
| } | ||
|
|
||
| /** | ||
| * Gets the address info assigned for this instance | ||
| */ | ||
| public get info(): AddressInfo { | ||
| return this.server.address() as AddressInfo; | ||
| } | ||
|
|
||
| /** | ||
| * Handles the incoming requests | ||
| * @param request | ||
| * @param response | ||
| * @private | ||
| */ | ||
| private async handle(request: Http.IncomingMessage, response: Http.ServerResponse): Promise<void> { | ||
| if (!request.url || request.method !== 'POST' && request.method !== 'DELETE') { | ||
| response.statusCode = 404; | ||
| response.statusMessage = 'Not Found'; | ||
| return void response.end(); | ||
| } | ||
|
|
||
| if (request.headers['authorization'] !== this.password) { | ||
| response.statusCode = 401; | ||
| response.statusMessage = 'Unauthorized'; | ||
| return void response.end(); | ||
| } | ||
|
|
||
| if (!request.url.includes('?shardId=')) { | ||
| response.statusCode = 400; | ||
| response.statusMessage = 'Bad Request'; | ||
| return void response.end('Missing shardId query string'); | ||
| } | ||
|
|
||
| const shardId = Number(request.url.split('?shardId=')[1]); | ||
|
|
||
| if (isNaN(shardId)) { | ||
| response.statusCode = 400; | ||
| response.statusMessage = 'Bad Request'; | ||
| return void response.end('Expected shardId to be a number'); | ||
| } | ||
|
|
||
| this.manager.emit(LibraryEvents.DEBUG, `Received a request in concurrency server! =>\n Url: ${request.url}\n Method: ${request.method}\n ShardId: ${shardId}`); | ||
|
|
||
| if (request.method === 'DELETE' && request.url.includes('/concurrency/cancel')) { | ||
| this.concurrency.abortIdentify(shardId); | ||
| response.statusCode = 200; | ||
| response.statusMessage = 'OK'; | ||
| return void response.end(); | ||
| } | ||
|
|
||
| if (request.method === 'POST' && request.url.includes('/concurrency/acquire')) { | ||
| try { | ||
| await this.concurrency.waitForIdentify(shardId); | ||
| response.statusCode = 204; | ||
| response.statusMessage = 'No Content'; | ||
| return void response.end(); | ||
| } catch (error) { | ||
| response.statusCode = 202; | ||
| response.statusMessage = 'Accepted'; | ||
| return void response.end('Acquire lock cancelled'); | ||
| } | ||
| } | ||
|
|
||
| response.statusCode = 404; | ||
| response.statusMessage = 'Not Found'; | ||
| return void response.end(); | ||
| } | ||
|
|
||
| /** | ||
| * Starts this server | ||
| */ | ||
| public start(): Promise<AddressInfo> { | ||
| return new Promise((resolve) => { | ||
| this.server.listen(0 , '127.0.0.1', () => resolve(this.info)); | ||
| }) | ||
| } | ||
| } |
Oops, something went wrong.