usbharu
/
misskey
mirror of https://github.com/misskey-dev/misskey.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Revert "enhance(backend): redisForJobQueue の接続を使い回す (MisskeyIO#268)" (MisskeyIO#292) 

This reverts commit 7122657f13.
This commit is contained in:
まっちゃとーにゅ 2023-12-28 05:33:22 +09:00 committed by GitHub
parent f207c962b5
commit bd265caf65
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 34 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
packages/backend/src/GlobalModule.ts
View File

@ -79,23 +79,11 @@ const $redisForTimelines: Provider = {
inject: [DI.config], inject: [DI.config],
}; };
const $redisForJobQueue: Provider = {
provide: DI.redisForJobQueue,
useFactory: (config: Config) => {
return new Redis.Redis({
...config.redisForJobQueue,
maxRetriesPerRequest: null,
keyPrefix: undefined,
});
},
inject: [DI.config],
};
@Global() @Global()
@Module({ @Module({
imports: [RepositoryModule], imports: [RepositoryModule],
providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForJobQueue], providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines],
exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForJobQueue, RepositoryModule], exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, RepositoryModule],
}) })
export class GlobalModule implements OnApplicationShutdown { export class GlobalModule implements OnApplicationShutdown {
constructor( constructor(
@ -104,7 +92,6 @@ export class GlobalModule implements OnApplicationShutdown {
@Inject(DI.redisForPub) private redisForPub: Redis.Redis, @Inject(DI.redisForPub) private redisForPub: Redis.Redis,
@Inject(DI.redisForSub) private redisForSub: Redis.Redis, @Inject(DI.redisForSub) private redisForSub: Redis.Redis,
@Inject(DI.redisForTimelines) private redisForTimelines: Redis.Redis, @Inject(DI.redisForTimelines) private redisForTimelines: Redis.Redis,
@Inject(DI.redisForJobQueue) private redisForJobQueue: Redis.Redis,
) {} ) {}
public async dispose(): Promise<void> { public async dispose(): Promise<void> {
@ -122,7 +109,6 @@ export class GlobalModule implements OnApplicationShutdown {
this.redisForPub.disconnect(), this.redisForPub.disconnect(),
this.redisForSub.disconnect(), this.redisForSub.disconnect(),
this.redisForTimelines.disconnect(), this.redisForTimelines.disconnect(),
this.redisForJobQueue.disconnect(),
]); ]);
} }

33
packages/backend/src/core/QueueModule.ts
View File

@ -6,7 +6,6 @@
import { setTimeout } from 'node:timers/promises'; import { setTimeout } from 'node:timers/promises';
import { Inject, Module, OnApplicationShutdown } from '@nestjs/common'; import { Inject, Module, OnApplicationShutdown } from '@nestjs/common';
import * as Bull from 'bullmq'; import * as Bull from 'bullmq';
import * as Redis from 'ioredis';
import { DI } from '@/di-symbols.js'; import { DI } from '@/di-symbols.js';
import type { Config } from '@/config.js'; import type { Config } from '@/config.js';
import { QUEUE, baseQueueOptions } from '@/queue/const.js'; import { QUEUE, baseQueueOptions } from '@/queue/const.js';
@ -24,50 +23,50 @@ export type WebhookDeliverQueue = Bull.Queue<WebhookDeliverJobData>;
const $system: Provider = { const $system: Provider = {
provide: 'queue:system', provide: 'queue:system',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.SYSTEM, baseQueueOptions(config, QUEUE.SYSTEM, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.SYSTEM, baseQueueOptions(config, QUEUE.SYSTEM)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
const $endedPollNotification: Provider = { const $endedPollNotification: Provider = {
provide: 'queue:endedPollNotification', provide: 'queue:endedPollNotification',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.ENDED_POLL_NOTIFICATION, baseQueueOptions(config, QUEUE.ENDED_POLL_NOTIFICATION, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.ENDED_POLL_NOTIFICATION, baseQueueOptions(config, QUEUE.ENDED_POLL_NOTIFICATION)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
const $deliver: Provider = { const $deliver: Provider = {
provide: 'queue:deliver', provide: 'queue:deliver',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
const $inbox: Provider = { const $inbox: Provider = {
provide: 'queue:inbox', provide: 'queue:inbox',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.INBOX, baseQueueOptions(config, QUEUE.INBOX, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.INBOX, baseQueueOptions(config, QUEUE.INBOX)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
const $db: Provider = { const $db: Provider = {
provide: 'queue:db', provide: 'queue:db',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.DB, baseQueueOptions(config, QUEUE.DB, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.DB, baseQueueOptions(config, QUEUE.DB)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
const $relationship: Provider = { const $relationship: Provider = {
provide: 'queue:relationship', provide: 'queue:relationship',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.RELATIONSHIP, baseQueueOptions(config, QUEUE.RELATIONSHIP, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.RELATIONSHIP, baseQueueOptions(config, QUEUE.RELATIONSHIP)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
const $objectStorage: Provider = { const $objectStorage: Provider = {
provide: 'queue:objectStorage', provide: 'queue:objectStorage',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.OBJECT_STORAGE, baseQueueOptions(config, QUEUE.OBJECT_STORAGE, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.OBJECT_STORAGE, baseQueueOptions(config, QUEUE.OBJECT_STORAGE)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
const $webhookDeliver: Provider = { const $webhookDeliver: Provider = {
provide: 'queue:webhookDeliver', provide: 'queue:webhookDeliver',
useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.WEBHOOK_DELIVER, redisForJobQueue)), useFactory: (config: Config) => new Bull.Queue(QUEUE.WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.WEBHOOK_DELIVER)),
inject: [DI.config, DI.redisForJobQueue], inject: [DI.config],
}; };
@Module({ @Module({

8
packages/backend/src/daemons/QueueStatsService.ts
View File

@ -6,7 +6,6 @@
import { Inject, Injectable } from '@nestjs/common'; import { Inject, Injectable } from '@nestjs/common';
import Xev from 'xev'; import Xev from 'xev';
import * as Bull from 'bullmq'; import * as Bull from 'bullmq';
import * as Redis from 'ioredis';
import { QueueService } from '@/core/QueueService.js'; import { QueueService } from '@/core/QueueService.js';
import { bindThis } from '@/decorators.js'; import { bindThis } from '@/decorators.js';
import { DI } from '@/di-symbols.js'; import { DI } from '@/di-symbols.js';
@ -26,9 +25,6 @@ export class QueueStatsService implements OnApplicationShutdown {
@Inject(DI.config) @Inject(DI.config)
private config: Config, private config: Config,
@Inject(DI.redisForJobQueue)
private redisForJobQueue: Redis.Redis,
private queueService: QueueService, private queueService: QueueService,
) { ) {
} }
@ -47,8 +43,8 @@ export class QueueStatsService implements OnApplicationShutdown {
let activeDeliverJobs = 0; let activeDeliverJobs = 0;
let activeInboxJobs = 0; let activeInboxJobs = 0;
const deliverQueueEvents = new Bull.QueueEvents(QUEUE.DELIVER, baseQueueOptions(this.config, QUEUE.DELIVER, this.redisForJobQueue)); const deliverQueueEvents = new Bull.QueueEvents(QUEUE.DELIVER, baseQueueOptions(this.config, QUEUE.DELIVER));
const inboxQueueEvents = new Bull.QueueEvents(QUEUE.INBOX, baseQueueOptions(this.config, QUEUE.INBOX, this.redisForJobQueue)); const inboxQueueEvents = new Bull.QueueEvents(QUEUE.INBOX, baseQueueOptions(this.config, QUEUE.INBOX));
deliverQueueEvents.on('active', () => { deliverQueueEvents.on('active', () => {
activeDeliverJobs++; activeDeliverJobs++;

1
packages/backend/src/di-symbols.ts
View File

@ -11,7 +11,6 @@ export const DI = {
redisForPub: Symbol('redisForPub'), redisForPub: Symbol('redisForPub'),
redisForSub: Symbol('redisForSub'), redisForSub: Symbol('redisForSub'),
redisForTimelines: Symbol('redisForTimelines'), redisForTimelines: Symbol('redisForTimelines'),
redisForJobQueue: Symbol('redisForJobQueue'),
//#region Repositories //#region Repositories
usersRepository: Symbol('usersRepository'), usersRepository: Symbol('usersRepository'),

20
packages/backend/src/queue/QueueProcessorService.ts
View File

@ -5,7 +5,6 @@
import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common'; import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common';
import * as Bull from 'bullmq'; import * as Bull from 'bullmq';
import * as Redis from 'ioredis';
import type { Config } from '@/config.js'; import type { Config } from '@/config.js';
import { DI } from '@/di-symbols.js'; import { DI } from '@/di-symbols.js';
import type Logger from '@/logger.js'; import type Logger from '@/logger.js';
@ -85,9 +84,6 @@ export class QueueProcessorService implements OnApplicationShutdown {
@Inject(DI.config) @Inject(DI.config)
private config: Config, private config: Config,
@Inject(DI.redisForJobQueue)
private redisForJobQueue: Redis.Redis,
private queueLoggerService: QueueLoggerService, private queueLoggerService: QueueLoggerService,
private webhookDeliverProcessorService: WebhookDeliverProcessorService, private webhookDeliverProcessorService: WebhookDeliverProcessorService,
private endedPollNotificationProcessorService: EndedPollNotificationProcessorService, private endedPollNotificationProcessorService: EndedPollNotificationProcessorService,
@ -150,7 +146,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
default: throw new Error(`unrecognized job type ${job.name} for system`); default: throw new Error(`unrecognized job type ${job.name} for system`);
} }
}, { }, {
...baseQueueOptions(this.config, QUEUE.SYSTEM, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.SYSTEM),
autorun: false, autorun: false,
}); });
@ -189,7 +185,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
default: throw new Error(`unrecognized job type ${job.name} for db`); default: throw new Error(`unrecognized job type ${job.name} for db`);
} }
}, { }, {
...baseQueueOptions(this.config, QUEUE.DB, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.DB),
autorun: false, autorun: false,
}); });
@ -205,7 +201,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
//#region deliver //#region deliver
this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => this.deliverProcessorService.process(job), { this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => this.deliverProcessorService.process(job), {
...baseQueueOptions(this.config, QUEUE.DELIVER, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.DELIVER),
autorun: false, autorun: false,
concurrency: this.config.deliverJobConcurrency ?? 128, concurrency: this.config.deliverJobConcurrency ?? 128,
limiter: { limiter: {
@ -229,7 +225,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
//#region inbox //#region inbox
this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => this.inboxProcessorService.process(job), { this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => this.inboxProcessorService.process(job), {
...baseQueueOptions(this.config, QUEUE.INBOX, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.INBOX),
autorun: false, autorun: false,
concurrency: this.config.inboxJobConcurrency ?? 16, concurrency: this.config.inboxJobConcurrency ?? 16,
limiter: { limiter: {
@ -253,7 +249,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
//#region webhook deliver //#region webhook deliver
this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => this.webhookDeliverProcessorService.process(job), { this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => this.webhookDeliverProcessorService.process(job), {
...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER),
autorun: false, autorun: false,
concurrency: 64, concurrency: 64,
limiter: { limiter: {
@ -285,7 +281,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
default: throw new Error(`unrecognized job type ${job.name} for relationship`); default: throw new Error(`unrecognized job type ${job.name} for relationship`);
} }
}, { }, {
...baseQueueOptions(this.config, QUEUE.RELATIONSHIP, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.RELATIONSHIP),
autorun: false, autorun: false,
concurrency: this.config.relashionshipJobConcurrency ?? 16, concurrency: this.config.relashionshipJobConcurrency ?? 16,
limiter: { limiter: {
@ -312,7 +308,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
default: throw new Error(`unrecognized job type ${job.name} for objectStorage`); default: throw new Error(`unrecognized job type ${job.name} for objectStorage`);
} }
}, { }, {
...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE),
autorun: false, autorun: false,
concurrency: 16, concurrency: 16,
}); });
@ -329,7 +325,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
//#region ended poll notification //#region ended poll notification
this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => this.endedPollNotificationProcessorService.process(job), { this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => this.endedPollNotificationProcessorService.process(job), {
...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION, this.redisForJobQueue), ...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
autorun: false, autorun: false,
}); });
//#endregion //#endregion

9
packages/backend/src/queue/const.ts
View File

@ -5,7 +5,6 @@
import { Config } from '@/config.js'; import { Config } from '@/config.js';
import type * as Bull from 'bullmq'; import type * as Bull from 'bullmq';
import type * as Redis from 'ioredis';
export const QUEUE = { export const QUEUE = {
DELIVER: 'deliver', DELIVER: 'deliver',
@ -18,9 +17,13 @@ export const QUEUE = {
WEBHOOK_DELIVER: 'webhookDeliver', WEBHOOK_DELIVER: 'webhookDeliver',
}; };
export function baseQueueOptions(config: Config, queueName: typeof QUEUE[keyof typeof QUEUE], redisConnection: Redis.Redis): Bull.QueueOptions { export function baseQueueOptions(config: Config, queueName: typeof QUEUE[keyof typeof QUEUE]): Bull.QueueOptions {
return { return {
connection: redisConnection, connection: {
...config.redisForJobQueue,
maxRetriesPerRequest: null,
keyPrefix: undefined,
},
prefix: config.redisForJobQueue.prefix ? `${config.redisForJobQueue.prefix}:queue:${queueName}` : `queue:${queueName}`, prefix: config.redisForJobQueue.prefix ? `${config.redisForJobQueue.prefix}:queue:${queueName}` : `queue:${queueName}`,
}; };
} }