diff --git a/.github/workflows/other-runtime-unit-test.yml b/.github/workflows/other-runtime-unit-test.yml index 6257a2f28..e580f66c1 100644 --- a/.github/workflows/other-runtime-unit-test.yml +++ b/.github/workflows/other-runtime-unit-test.yml @@ -20,7 +20,7 @@ jobs: cache: yarn - uses: denoland/setup-deno@v2 with: - deno-version: 'v2.1.x' + deno-version: 'v2.9.x' - run: yarn install --immutable - name: Cache for Turbo uses: rharkor/caching-for-turbo@v2.5.0 @@ -39,7 +39,7 @@ jobs: cache: yarn - uses: oven-sh/setup-bun@v2 with: - bun-version: '1.1.42' + bun-version: '1.3.14' - run: yarn install --immutable - name: Cache for Turbo uses: rharkor/caching-for-turbo@v2.5.0 diff --git a/packages/gateway/package.json b/packages/gateway/package.json index 9104b184c..570acdc5d 100644 --- a/packages/gateway/package.json +++ b/packages/gateway/package.json @@ -29,9 +29,6 @@ "@discordeno/utils": "workspace:^", "ws": "^8.21.0" }, - "optionalDependencies": { - "fzstd": "^0.1.1" - }, "devDependencies": { "@biomejs/biome": "2.5.5", "@swc/cli": "^0.8.1", diff --git a/packages/gateway/src/Shard.ts b/packages/gateway/src/Shard.ts index f3df36698..09dddee07 100644 --- a/packages/gateway/src/Shard.ts +++ b/packages/gateway/src/Shard.ts @@ -1,9 +1,8 @@ import { Buffer } from 'node:buffer'; -import zlib from 'node:zlib'; +import { createInflate, createZstdDecompress, type Inflate, inflateSync, type ZstdDecompress, constants as zlibConstants } from 'node:zlib'; import type { DiscordGatewayPayload, DiscordHello, DiscordReady, DiscordUpdatePresence } from '@discordeno/types'; import { GatewayCloseEventCodes, GatewayOpcodes } from '@discordeno/types'; import { delay, LeakyBucket, logger } from '@discordeno/utils'; -import type { Decompress as FZstdDecompress } from 'fzstd'; import NodeWebSocket from 'ws'; import { type ShardEvents, @@ -17,13 +16,6 @@ import { const ZLIB_SYNC_FLUSH = new Uint8Array([0x0, 0x0, 0xff, 0xff]); -let fzstd: typeof import('fzstd'); - -/** Since fzstd is an optional dependency, we need to import it lazily. */ -async function getFZStd() { - return (fzstd ??= await import('fzstd')); -} - export class DiscordenoShard { /** The id of the shard. */ id: number; @@ -67,14 +59,10 @@ export class DiscordenoShard { goingOffline = false; /** Text decoder used for compressed payloads. */ textDecoder = new TextDecoder(); - /** zlib Inflate or zstd decompress (from node:zlib) instance for transport payloads. */ - inflate?: zlib.Inflate | zlib.ZstdDecompress; - /** ZLib inflate buffer. */ + /** zlib Inflate or zstd decompress instance for transport payloads. */ + inflate?: Inflate | ZstdDecompress; + /** zlib inflate/zstd buffer. */ inflateBuffer: Uint8Array | null = null; - /** ZStd Decompress instance for ZStd-stream transport payloads. */ - zstdDecompress?: FZstdDecompress; - /** Queue for compressed payloads for Zstd Decompress */ - decompressionPromisesQueue: ((data: DiscordGatewayPayload) => void)[] = []; /** * A function that will be called once the socket is closed and handleClose() has finished updating internal states. * @@ -183,79 +171,39 @@ export class DiscordenoShard { if (this.gatewayConfig.transportCompression) { url.searchParams.set('compress', this.gatewayConfig.transportCompression); + this.inflateBuffer = null; + if (this.gatewayConfig.transportCompression === TransportCompression.zlib) { - this.inflateBuffer = null; - this.inflate = zlib.createInflate({ - finishFlush: zlib.constants.Z_SYNC_FLUSH, + this.inflate = createInflate({ + finishFlush: zlibConstants.Z_SYNC_FLUSH, chunkSize: 64 * 1024, }); - - this.inflate.on('error', (e) => { - this.logger.error('The was an error in decompressing a ZLib compressed payload', e); - }); - - this.inflate.on('data', (data) => { - if (!(data instanceof Uint8Array)) return; - - if (this.inflateBuffer) { - const newBuffer = new Uint8Array(this.inflateBuffer.byteLength + data.byteLength); - newBuffer.set(this.inflateBuffer); - newBuffer.set(data, this.inflateBuffer.byteLength); - this.inflateBuffer = newBuffer; - - return; - } - - this.inflateBuffer = data; + } else if (this.gatewayConfig.transportCompression === TransportCompression.zstd) { + this.inflate = createZstdDecompress({ + chunkSize: 64 * 1024, }); + } else { + throw new Error(`[Shard] Unknown transport compression type: ${this.gatewayConfig.transportCompression}`); } - if (this.gatewayConfig.transportCompression === TransportCompression.zstd) { - if ('createZstdDecompress' in zlib) { - this.logger.debug('[Shard] Using node:zlib zstd decompression.'); + this.inflate.on('error', (e) => { + this.logger.error('The was an error in decompressing a compressed payload', e); + }); - this.inflateBuffer = null; - this.inflate = zlib.createZstdDecompress({ - chunkSize: 64 * 1024, - }); + this.inflate.on('data', (data) => { + if (!(data instanceof Uint8Array)) return; - this.inflate.on('error', (e) => { - this.logger.error('The was an error in decompressing a Zstd compressed payload', e); - }); + if (this.inflateBuffer) { + const newBuffer = new Uint8Array(this.inflateBuffer.byteLength + data.byteLength); + newBuffer.set(this.inflateBuffer); + newBuffer.set(data, this.inflateBuffer.byteLength); + this.inflateBuffer = newBuffer; - this.inflate.on('data', (data) => { - if (!(data instanceof Uint8Array)) return; - - if (this.inflateBuffer) { - const newBuffer = new Uint8Array(this.inflateBuffer.byteLength + data.byteLength); - newBuffer.set(this.inflateBuffer); - newBuffer.set(data, this.inflateBuffer.byteLength); - this.inflateBuffer = newBuffer; - - return; - } - - this.inflateBuffer = data; - }); - } else { - const fzstd = await getFZStd().catch(() => { - this.logger.warn('[Shard] "fzstd" is not installed. Disabled transport compression.'); - url.searchParams.delete('compress'); - - return null; - }); - - if (fzstd) { - this.logger.debug('[Shard] Using fzstd zstd decompression.'); - - this.zstdDecompress = new fzstd.Decompress((data) => { - const decodedData = this.textDecoder.decode(data); - const parsedData = JSON.parse(decodedData); - this.decompressionPromisesQueue.shift()?.(parsedData); - }); - } + return; } - } + + this.inflateBuffer = data; + }); } if (this.gatewayConfig.compress && this.gatewayConfig.transportCompression) { @@ -439,9 +387,7 @@ export class DiscordenoShard { // Clear the zlib/zstd data this.inflate = undefined; - this.zstdDecompress = undefined; this.inflateBuffer = null; - this.decompressionPromisesQueue = []; this.logger.debug(`[Shard] Shard #${this.id} closed with code ${close.code}${close.reason ? `, and reason: ${close.reason}` : ''}.`); @@ -548,20 +494,17 @@ export class DiscordenoShard { // A buffer is a Uint8Array under the hood. An ArrayBuffer is generic, so we need to create the Uint8Array that uses the whole ArrayBuffer const compressedData: Uint8Array = data instanceof Buffer ? data : new Uint8Array(data); - if (this.gatewayConfig.transportCompression === TransportCompression.zlib) { + if (this.gatewayConfig.transportCompression) { if (!this.inflate) { - this.logger.fatal('[Shard] zlib-stream transport compression was enabled but no instance of Inflate was found.'); + this.logger.fatal('[Shard] Transport compression was enabled but no instance of the decompression was found.'); return null; } - // Alias, used to avoid some null checks in the Promise constructor - const inflate = this.inflate; - const writePromise = new Promise((resolve, reject) => { - inflate.write(compressedData, 'binary', (error) => (error ? reject(error) : resolve())); + this.inflate!.write(compressedData, 'binary', (error) => (error ? reject(error) : resolve())); }); - if (!endsWithMarker(compressedData, ZLIB_SYNC_FLUSH)) return null; + if (this.gatewayConfig.transportCompression === TransportCompression.zlib && !endsWithMarker(compressedData, ZLIB_SYNC_FLUSH)) return null; await writePromise; @@ -576,41 +519,8 @@ export class DiscordenoShard { return JSON.parse(decodedData); } - if (this.gatewayConfig.transportCompression === TransportCompression.zstd) { - if (this.zstdDecompress) { - this.zstdDecompress.push(compressedData); - - const decompressionPromise = new Promise((r) => this.decompressionPromisesQueue.push(r)); - return await decompressionPromise; - } - - if (this.inflate) { - // Alias, used to avoid some null checks in the Promise constructor - const decompress = this.inflate; - - const writePromise = new Promise((resolve, reject) => { - decompress.write(compressedData, 'binary', (error) => (error ? reject(error) : resolve())); - }); - - await writePromise; - - if (!this.inflateBuffer) { - this.logger.warn('[Shard] The ZLib inflate buffer was cleared at an unexpected moment.'); - return null; - } - - const decodedData = this.textDecoder.decode(this.inflateBuffer); - this.inflateBuffer = null; - - return JSON.parse(decodedData); - } - - this.logger.fatal('[Shard] zstd-stream transport compression was enabled but no zstd decompressor was found.'); - return null; - } - if (this.gatewayConfig.compress) { - const decompressed = zlib.inflateSync(compressedData); + const decompressed = inflateSync(compressedData); const decodedData = this.textDecoder.decode(decompressed); return JSON.parse(decodedData); diff --git a/yarn.lock b/yarn.lock index ec58c1e83..18fd56bf6 100644 --- a/yarn.lock +++ b/yarn.lock @@ -160,16 +160,12 @@ __metadata: "@types/ws": "npm:^8.18.1" c8: "npm:^11.0.0" chai: "npm:^6.2.2" - fzstd: "npm:^0.1.1" mocha: "npm:^11.7.6" sinon: "npm:^22.0.0" ts-node: "npm:^10.9.2" tsconfig: "npm:*" typescript: "npm:^6.0.3" ws: "npm:^8.21.0" - dependenciesMeta: - fzstd: - optional: true languageName: unknown linkType: soft @@ -1821,13 +1817,6 @@ __metadata: languageName: node linkType: hard -"fzstd@npm:^0.1.1": - version: 0.1.1 - resolution: "fzstd@npm:0.1.1" - checksum: 10c0/c4ae25a4b9e7ac58e9716fcb0c4ec19f3e678d90a67a59e97ce361beb27e4aa4b8666b58b4dc2d715e92db4b65b189c978186a5fedba6fccb5ec9765717dd1cd - languageName: node - linkType: hard - "get-caller-file@npm:^2.0.5": version: 2.0.5 resolution: "get-caller-file@npm:2.0.5"