From f930eaee020520c1f9496803dcf445b2f2955526 Mon Sep 17 00:00:00 2001 From: syuilo Date: Mon, 29 May 2023 13:32:19 +0900 Subject: [PATCH] perf(backend): use websockets/ws instead of theturtle32/WebSocket-Node (#10884) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * perf(backend): use websockets/ws instead of theturtle32/WebSocket-Node Resolve #10883 * refactor * Update StreamingApiServerService.ts * Update StreamingApiServerService.ts * :v: * Update StreamingApiServerService.ts * fix main stream init * fix timing 2 * setIntervalの重複を避ける（気休め） * add comment * :v: --------- Co-authored-by: tamaina --- packages/backend/package.json | 5 +- packages/backend/src/server/ServerService.ts | 3 +- .../src/server/api/AuthenticateService.ts | 2 +- .../server/api/StreamingApiServerService.ts | 121 ++++++++++-------- .../backend/src/server/api/stream/index.ts | 21 ++- pnpm-lock.yaml | 63 +++------ 6 files changed, 107 insertions(+), 108 deletions(-) diff --git a/packages/backend/package.json b/packages/backend/package.json index 99c04d6bf..66bc7ee47 100644 --- a/packages/backend/package.json +++ b/packages/backend/package.json @@ -35,6 +35,7 @@ "@swc/core-win32-x64-msvc": "1.3.56", "@tensorflow/tfjs": "4.4.0", "@tensorflow/tfjs-node": "4.4.0", + "bufferutil": "^4.0.7", "slacc-android-arm-eabi": "0.0.9", "slacc-android-arm64": "0.0.9", "slacc-darwin-arm64": "0.0.9", @@ -46,7 +47,8 @@ "slacc-linux-arm64-musl": "0.0.9", "slacc-linux-x64-gnu": "0.0.9", "slacc-win32-arm64-msvc": "0.0.9", - "slacc-win32-x64-msvc": "0.0.9" + "slacc-win32-x64-msvc": "0.0.9", + "utf-8-validate": "^6.0.3" }, "dependencies": { "@aws-sdk/client-s3": "3.321.1", @@ -157,7 +159,6 @@ "uuid": "9.0.0", "vary": "1.1.2", "web-push": "3.6.1", - "websocket": "1.0.34", "ws": "8.13.0", "xev": "3.0.2" }, diff --git a/packages/backend/src/server/ServerService.ts b/packages/backend/src/server/ServerService.ts index ce6a1f704..c3d45e4ad 100644 --- a/packages/backend/src/server/ServerService.ts +++ b/packages/backend/src/server/ServerService.ts @@ -194,7 +194,7 @@ export class ServerService implements OnApplicationShutdown { fastify.register(this.clientServerService.createServer); - this.streamingApiServerService.attachStreamingApi(fastify.server); + this.streamingApiServerService.attach(fastify.server); fastify.server.on('error', err => { switch ((err as any).code) { @@ -224,6 +224,7 @@ export class ServerService implements OnApplicationShutdown { @bindThis public async dispose(): Promise { + await this.streamingApiServerService.detach(); await this.#fastify.close(); } diff --git a/packages/backend/src/server/api/AuthenticateService.ts b/packages/backend/src/server/api/AuthenticateService.ts index 6548c475b..e23591d87 100644 --- a/packages/backend/src/server/api/AuthenticateService.ts +++ b/packages/backend/src/server/api/AuthenticateService.ts @@ -36,7 +36,7 @@ export class AuthenticateService { } @bindThis - public async authenticate(token: string | null | undefined): Promise<[LocalUser | null | undefined, AccessToken | null | undefined]> { + public async authenticate(token: string | null | undefined): Promise<[LocalUser | null, AccessToken | null]> { if (token == null) { return [null, null]; } diff --git a/packages/backend/src/server/api/StreamingApiServerService.ts b/packages/backend/src/server/api/StreamingApiServerService.ts index 258e8de03..fdda581ad 100644 --- a/packages/backend/src/server/api/StreamingApiServerService.ts +++ b/packages/backend/src/server/api/StreamingApiServerService.ts @@ -1,23 +1,25 @@ import { EventEmitter } from 'events'; import { Inject, Injectable } from '@nestjs/common'; import * as Redis from 'ioredis'; -import * as websocket from 'websocket'; +import * as WebSocket from 'ws'; import { DI } from '@/di-symbols.js'; -import type { UsersRepository, BlockingsRepository, ChannelFollowingsRepository, FollowingsRepository, MutingsRepository, UserProfilesRepository, RenoteMutingsRepository } from '@/models/index.js'; +import type { UsersRepository, AccessToken } from '@/models/index.js'; import type { Config } from '@/config.js'; import { NoteReadService } from '@/core/NoteReadService.js'; import { GlobalEventService } from '@/core/GlobalEventService.js'; import { NotificationService } from '@/core/NotificationService.js'; import { bindThis } from '@/decorators.js'; import { CacheService } from '@/core/CacheService.js'; -import { AuthenticateService } from './AuthenticateService.js'; +import { LocalUser } from '@/models/entities/User'; +import { AuthenticateService, AuthenticationError } from './AuthenticateService.js'; import MainStreamConnection from './stream/index.js'; import { ChannelsService } from './stream/ChannelsService.js'; -import type { ParsedUrlQuery } from 'querystring'; import type * as http from 'node:http'; @Injectable() export class StreamingApiServerService { + #wss: WebSocket.WebSocketServer; + constructor( @Inject(DI.config) private config: Config, @@ -28,24 +30,6 @@ export class StreamingApiServerService { @Inject(DI.usersRepository) private usersRepository: UsersRepository, - @Inject(DI.followingsRepository) - private followingsRepository: FollowingsRepository, - - @Inject(DI.mutingsRepository) - private mutingsRepository: MutingsRepository, - - @Inject(DI.renoteMutingsRepository) - private renoteMutingsRepository: RenoteMutingsRepository, - - @Inject(DI.blockingsRepository) - private blockingsRepository: BlockingsRepository, - - @Inject(DI.channelFollowingsRepository) - private channelFollowingsRepository: ChannelFollowingsRepository, - - @Inject(DI.userProfilesRepository) - private userProfilesRepository: UserProfilesRepository, - private cacheService: CacheService, private noteReadService: NoteReadService, private authenticateService: AuthenticateService, @@ -55,25 +39,65 @@ export class StreamingApiServerService { } @bindThis - public attachStreamingApi(server: http.Server) { - // Init websocket server - const ws = new websocket.server({ - httpServer: server, + public attach(server: http.Server): void { + this.#wss = new WebSocket.WebSocketServer({ + noServer: true, }); - ws.on('request', async (request) => { - const q = request.resourceURL.query as ParsedUrlQuery; - - // TODO: トークンが間違ってるなどしてauthenticateに失敗したら - // コネクション切断するなりエラーメッセージ返すなりする - // (現状はエラーがキャッチされておらずサーバーのログに流れて邪魔なので) - const [user, miapp] = await this.authenticateService.authenticate(q.i as string); - - if (user?.isSuspended) { - request.reject(400); + server.on('upgrade', async (request, socket, head) => { + if (request.url == null) { + socket.write('HTTP/1.1 400 Bad Request\r\n\r\n'); + socket.destroy(); return; } + const q = new URL(request.url, `http://${request.headers.host}`).searchParams; + + let user: LocalUser | null = null; + let app: AccessToken | null = null; + + try { + [user, app] = await this.authenticateService.authenticate(q.get('i')); + } catch (e) { + if (e instanceof AuthenticationError) { + socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n'); + } else { + socket.write('HTTP/1.1 500 Internal Server Error\r\n\r\n'); + } + socket.destroy(); + return; + } + + if (user?.isSuspended) { + socket.write('HTTP/1.1 403 Forbidden\r\n\r\n'); + socket.destroy(); + return; + } + + const stream = new MainStreamConnection( + this.channelsService, + this.noteReadService, + this.notificationService, + this.cacheService, + user, app, + ); + + await stream.init(); + + this.#wss.handleUpgrade(request, socket, head, (ws) => { + this.#wss.emit('connection', ws, request, { + stream, user, app, + }); + }); + }); + + this.#wss.on('connection', async (connection: WebSocket.WebSocket, request: http.IncomingMessage, ctx: { + stream: MainStreamConnection, + user: LocalUser | null; + app: AccessToken | null + }) => { + const { stream, user, app } = ctx; + const ev = new EventEmitter(); async function onRedisMessage(_: string, data: string): Promise { @@ -83,19 +107,7 @@ export class StreamingApiServerService { this.redisForSub.on('message', onRedisMessage); - const main = new MainStreamConnection( - this.channelsService, - this.noteReadService, - this.notificationService, - this.cacheService, - ev, user, miapp, - ); - - await main.init(); - - const connection = request.accept(); - - main.init2(connection); + await stream.listen(ev, connection); const intervalId = user ? setInterval(() => { this.usersRepository.update(user.id, { @@ -110,16 +122,23 @@ export class StreamingApiServerService { connection.once('close', () => { ev.removeAllListeners(); - main.dispose(); + stream.dispose(); this.redisForSub.off('message', onRedisMessage); if (intervalId) clearInterval(intervalId); }); connection.on('message', async (data) => { - if (data.type === 'utf8' && data.utf8Data === 'ping') { + if (data.toString() === 'ping') { connection.send('pong'); } }); }); } + + @bindThis + public detach(): Promise { + return new Promise((resolve) => { + this.#wss.close(() => resolve()); + }); + } } diff --git a/packages/backend/src/server/api/stream/index.ts b/packages/backend/src/server/api/stream/index.ts index fee56e366..8b1c2c09c 100644 --- a/packages/backend/src/server/api/stream/index.ts +++ b/packages/backend/src/server/api/stream/index.ts @@ -1,3 +1,4 @@ +import * as WebSocket from 'ws'; import type { User } from '@/models/entities/User.js'; import type { AccessToken } from '@/models/entities/AccessToken.js'; import type { Packed } from '@/misc/json-schema.js'; @@ -7,7 +8,6 @@ import { bindThis } from '@/decorators.js'; import { CacheService } from '@/core/CacheService.js'; import { UserProfile } from '@/models/index.js'; import type { ChannelsService } from './ChannelsService.js'; -import type * as websocket from 'websocket'; import type { EventEmitter } from 'events'; import type Channel from './channel.js'; import type { StreamEventEmitter, StreamMessages } from './types.js'; @@ -18,7 +18,7 @@ import type { StreamEventEmitter, StreamMessages } from './types.js'; export default class Connection { public user?: User; public token?: AccessToken; - private wsConnection: websocket.connection; + private wsConnection: WebSocket.WebSocket; public subscriber: StreamEventEmitter; private channels: Channel[] = []; private subscribingNotes: any = {}; @@ -37,11 +37,9 @@ export default class Connection { private notificationService: NotificationService, private cacheService: CacheService, - subscriber: EventEmitter, user: User | null | undefined, token: AccessToken | null | undefined, ) { - this.subscriber = subscriber; if (user) this.user = user; if (token) this.token = token; } @@ -70,12 +68,16 @@ export default class Connection { if (this.user != null) { await this.fetch(); - this.fetchIntervalId = setInterval(this.fetch, 1000 * 10); + if (!this.fetchIntervalId) { + this.fetchIntervalId = setInterval(this.fetch, 1000 * 10); + } } } @bindThis - public async init2(wsConnection: websocket.connection) { + public async listen(subscriber: EventEmitter, wsConnection: WebSocket.WebSocket) { + this.subscriber = subscriber; + this.wsConnection = wsConnection; this.wsConnection.on('message', this.onWsConnectionMessage); @@ -88,14 +90,11 @@ export default class Connection { * クライアントからメッセージ受信時 */ @bindThis - private async onWsConnectionMessage(data: websocket.Message) { - if (data.type !== 'utf8') return; - if (data.utf8Data == null) return; - + private async onWsConnectionMessage(data: WebSocket.RawData) { let obj: Record; try { - obj = JSON.parse(data.utf8Data); + obj = JSON.parse(data.toString()); } catch (e) { return; } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7930458cb..57e8cf850 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -96,7 +96,7 @@ importers: version: 8.2.1 '@fastify/http-proxy': specifier: 9.1.0 - version: 9.1.0 + version: 9.1.0(bufferutil@4.0.7)(utf-8-validate@6.0.3) '@fastify/multipart': specifier: 7.6.0 version: 7.6.0 @@ -219,7 +219,7 @@ importers: version: 4.1.0 jsdom: specifier: 21.1.1 - version: 21.1.1 + version: 21.1.1(bufferutil@4.0.7)(utf-8-validate@6.0.3) json5: specifier: 2.2.3 version: 2.2.3 @@ -388,12 +388,9 @@ importers: web-push: specifier: 3.6.1 version: 3.6.1 - websocket: - specifier: 1.0.34 - version: 1.0.34 ws: specifier: 8.13.0 - version: 8.13.0 + version: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3) xev: specifier: 3.0.2 version: 3.0.2 @@ -437,6 +434,9 @@ importers: '@tensorflow/tfjs-node': specifier: 4.4.0 version: 4.4.0(seedrandom@3.0.5) + bufferutil: + specifier: ^4.0.7 + version: 4.0.7 slacc-android-arm-eabi: specifier: 0.0.9 version: 0.0.9 @@ -473,6 +473,9 @@ importers: slacc-win32-x64-msvc: specifier: 0.0.9 version: 0.0.9 + utf-8-validate: + specifier: ^6.0.3 + version: 6.0.3 devDependencies: '@jest/globals': specifier: 29.5.0 @@ -3852,12 +3855,12 @@ packages: fast-json-stringify: 5.7.0 dev: false - /@fastify/http-proxy@9.1.0: + /@fastify/http-proxy@9.1.0(bufferutil@4.0.7)(utf-8-validate@6.0.3): resolution: {integrity: sha512-vgHCTDKOqLB437zQJiLWFFnsrYfFZ6Lfwu/xXQoKqRUKIPDt+xG6LBRtf8s5MNqfFVoTE7kw1U/0qdRGDsMp4Q==} dependencies: '@fastify/reply-from': 9.0.1 fastify-plugin: 4.5.0 - ws: 8.13.0 + ws: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3) transitivePeerDependencies: - bufferutil - utf-8-validate @@ -5532,7 +5535,7 @@ packages: ts-dedent: 2.2.0 util-deprecate: 1.0.2 watchpack: 2.4.0 - ws: 8.13.0 + ws: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3) transitivePeerDependencies: - bufferutil - encoding @@ -8702,7 +8705,6 @@ packages: requiresBuild: true dependencies: node-gyp-build: 4.6.0 - dev: false /bullmq@3.14.1: resolution: {integrity: sha512-Fom78UKljYsnJmwbROVPx3eFLuVfQjQbw9KCnVupLzT31RQHhFHV2xd/4J4oWl4u34bZ1JmEUfNnqNBz+IOJuA==} @@ -13872,7 +13874,7 @@ packages: - supports-color dev: true - /jsdom@21.1.1: + /jsdom@21.1.1(bufferutil@4.0.7)(utf-8-validate@6.0.3): resolution: {integrity: sha512-Jjgdmw48RKcdAIQyUD1UdBh2ecH7VqwaXPN3ehoZN6MqgVbMn+lRm1aAT1AsdJRAJpwfa4IpwgzySn61h2qu3w==} engines: {node: '>=14'} peerDependencies: @@ -13905,7 +13907,7 @@ packages: whatwg-encoding: 2.0.0 whatwg-mimetype: 3.0.0 whatwg-url: 12.0.1 - ws: 8.13.0 + ws: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3) xml-name-validator: 4.0.0 transitivePeerDependencies: - bufferutil @@ -15231,7 +15233,7 @@ packages: /node-gyp-build@4.6.0: resolution: {integrity: sha512-NTZVKn9IylLwUzaKjkas1e4u2DLNcV4rdYagA4PWdPwW87Bi7z+BznyKSRwS/761tV/lzCGXplWsiaMjLqP2zQ==} - dev: false + hasBin: true /node-gyp@9.3.1: resolution: {integrity: sha512-4Q16ZCqq3g8awk6UplT7AuxQ35XN4R/yf/+wSAwcBUAjg7l58RTactWaP8fIDTi0FzI7YcVLujwExakZlfWkXg==} @@ -19276,12 +19278,6 @@ packages: resolution: {integrity: sha512-dzlvlNlt6AXU7EBSfpAscydQ7gXB+pPGsPnfJnZpiNJBDj7IaJzQlBZYGdEi4R9HmPdBv2XmWJ6YUtoTa7lmCw==} dev: false - /typedarray-to-buffer@3.1.5: - resolution: {integrity: sha512-zdu8XMNEDepKKR+XYOXAVPtWui0ly0NtohUscw+UmaHiAWT8hrV1rr//H6V+0DvJ3OQ19S979M0laLfX8rm82Q==} - dependencies: - is-typedarray: 1.0.0 - dev: false - /typedarray@0.0.6: resolution: {integrity: sha512-/aCDEGatGvZ2BIk+HmLf4ifCJFwvKFNb9/JeZPMulfgFracn9QFcAf5GO8B/mweUjSoblS5In0cWhqpfs/5PQA==} @@ -19656,13 +19652,12 @@ packages: resolution: {integrity: sha512-cwESVXlO3url9YWlFW/TA9cshCEhtu7IKJ/p5soJ/gGpj7vbvFrAY/eIioQ6Dw23KjZhYgiIo8HOs1nQ2vr/oQ==} engines: {node: '>=0.10.0'} - /utf-8-validate@5.0.10: - resolution: {integrity: sha512-Z6czzLq4u8fPOyx7TU6X3dvUZVvoJmxSQ+IcrlmagKhilxlhZgxPK6C5Jqbkw1IDUmFTM+cz9QDnnLTwDz/2gQ==} + /utf-8-validate@6.0.3: + resolution: {integrity: sha512-uIuGf9TWQ/y+0Lp+KGZCMuJWc3N9BHA+l/UmHd/oUHwJJDeysyTRxNQVkbzsIWfGFbRe3OcgML/i0mvVRPOyDA==} engines: {node: '>=6.14.2'} requiresBuild: true dependencies: node-gyp-build: 4.6.0 - dev: false /util-deprecate@1.0.2: resolution: {integrity: sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==} @@ -20132,20 +20127,6 @@ packages: resolution: {integrity: sha512-kyDivFZ7ZM0BVOUteVbDFhlRt7Ah/CSPwJdi8hBpkK7QLumUqdLtVfm/PX/hkcnrvr0i77fO5+TjZ94Pe+C9iw==} dev: false - /websocket@1.0.34: - resolution: {integrity: sha512-PRDso2sGwF6kM75QykIesBijKSVceR6jL2G8NGYyq2XrItNC2P5/qL5XeR056GhA+Ly7JMFvJb9I312mJfmqnQ==} - engines: {node: '>=4.0.0'} - dependencies: - bufferutil: 4.0.7 - debug: 2.6.9 - es5-ext: 0.10.62 - typedarray-to-buffer: 3.1.5 - utf-8-validate: 5.0.10 - yaeti: 0.0.6 - transitivePeerDependencies: - - supports-color - dev: false - /well-known-symbols@2.0.0: resolution: {integrity: sha512-ZMjC3ho+KXo0BfJb7JgtQ5IBuvnShdlACNkKkdsqBmYw3bPAaJfPeYUo6tLUaT5tG/Gkh7xkpBhKRQ9e7pyg9Q==} engines: {node: '>=6'} @@ -20336,7 +20317,7 @@ packages: async-limiter: 1.0.1 dev: true - /ws@8.13.0: + /ws@8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3): resolution: {integrity: sha512-x9vcZYTrFPC7aSIbj7sRCYo7L/Xb8Iy+pW0ng0wt2vCJv7M9HOMy0UoN3rr+IFC7hb7vXoqS+P9ktyLLLhO+LA==} engines: {node: '>=10.0.0'} peerDependencies: @@ -20347,6 +20328,9 @@ packages: optional: true utf-8-validate: optional: true + dependencies: + bufferutil: 4.0.7 + utf-8-validate: 6.0.3 /xev@3.0.2: resolution: {integrity: sha512-8kxuH95iMXzHZj+fwqfA4UrPcYOy6bGIgfWzo9Ji23JoEc30ge/Z++Ubkiuy8c0+M64nXmmxrmJ7C8wnuBhluw==} @@ -20394,11 +20378,6 @@ packages: resolution: {integrity: sha512-0pfFzegeDWJHJIAmTLRP2DwHjdF5s7jo9tuztdQxAhINCdvS+3nGINqPd00AphqJR/0LhANUS6/+7SCb98YOfA==} engines: {node: '>=10'} - /yaeti@0.0.6: - resolution: {integrity: sha512-MvQa//+KcZCUkBTIC9blM+CU9J2GzuTytsOUwf2lidtvkx/6gnEp1QvJv34t9vdjhFmha/mUiNDbN0D0mJWdug==} - engines: {node: '>=0.10.32'} - dev: false - /yallist@2.1.2: resolution: {integrity: sha512-ncTzHV7NvsQZkYe1DW7cbDLm0YpzHmZF5r/iyP3ZnQtMiJ+pjzisCiMNI+Sj+xQF5pXhSHxSB3uDbsBTzY/c2A==}