Mizah/mizzkey
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

spec(Queue): lockDurationとstalledIntervalを増やす (MisskeyIO#308)

Browse source
This commit is contained in:
まっちゃとーにゅ 2024-01-01 02:57:23 +09:00 committed by GitHub
parent efee8f45f9
commit e7643da3bd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 24 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
packages/backend/src/queue/QueueProcessorService.ts
View file

@ -40,7 +40,7 @@ import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMu
import { CleanProcessorService } from './processors/CleanProcessorService.js'; import { CleanProcessorService } from './processors/CleanProcessorService.js';
import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js'; import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js';
import { QueueLoggerService } from './QueueLoggerService.js'; import { QueueLoggerService } from './QueueLoggerService.js';
import { QUEUE, baseQueueOptions } from './const.js'; import { QUEUE, baseWorkerOptions } from './const.js';
// ref. https://github.com/misskey-dev/misskey/pull/7635#issue-971097019 // ref. https://github.com/misskey-dev/misskey/pull/7635#issue-971097019
function httpRelatedBackoff(attemptsMade: number) { function httpRelatedBackoff(attemptsMade: number) {
@ -146,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.redisForSystemQueue, QUEUE.SYSTEM), ...baseWorkerOptions(this.config.redisForSystemQueue, QUEUE.SYSTEM),
autorun: false, autorun: false,
}); });
@ -185,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.redisForDbQueue, QUEUE.DB), ...baseWorkerOptions(this.config.redisForDbQueue, QUEUE.DB),
autorun: false, autorun: false,
}); });
@ -201,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.redisForDeliverQueue, QUEUE.DELIVER), ...baseWorkerOptions(this.config.redisForDeliverQueue, QUEUE.DELIVER),
autorun: false, autorun: false,
concurrency: this.config.deliverJobConcurrency ?? 128, concurrency: this.config.deliverJobConcurrency ?? 128,
limiter: { limiter: {
@ -225,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.redisForInboxQueue, QUEUE.INBOX), ...baseWorkerOptions(this.config.redisForInboxQueue, QUEUE.INBOX),
autorun: false, autorun: false,
concurrency: this.config.inboxJobConcurrency ?? 16, concurrency: this.config.inboxJobConcurrency ?? 16,
limiter: { limiter: {
@ -249,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.redisForWebhookDeliverQueue, QUEUE.WEBHOOK_DELIVER), ...baseWorkerOptions(this.config.redisForWebhookDeliverQueue, QUEUE.WEBHOOK_DELIVER),
autorun: false, autorun: false,
concurrency: 64, concurrency: 64,
limiter: { limiter: {
@ -281,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.redisForRelationshipQueue, QUEUE.RELATIONSHIP), ...baseWorkerOptions(this.config.redisForRelationshipQueue, QUEUE.RELATIONSHIP),
autorun: false, autorun: false,
concurrency: this.config.relashionshipJobConcurrency ?? 16, concurrency: this.config.relashionshipJobConcurrency ?? 16,
limiter: { limiter: {
@ -308,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.redisForObjectStorageQueue, QUEUE.OBJECT_STORAGE), ...baseWorkerOptions(this.config.redisForObjectStorageQueue, QUEUE.OBJECT_STORAGE),
autorun: false, autorun: false,
concurrency: 16, concurrency: 16,
}); });
@ -325,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.redisForEndedPollNotificationQueue, QUEUE.ENDED_POLL_NOTIFICATION), ...baseWorkerOptions(this.config.redisForEndedPollNotificationQueue, QUEUE.ENDED_POLL_NOTIFICATION),
autorun: false, autorun: false,
}); });
//#endregion //#endregion

15
packages/backend/src/queue/const.ts
View file

@ -28,3 +28,18 @@ export function baseQueueOptions(config: RedisOptions & RedisOptionsSource, queu
prefix: config.prefix ? `${config.prefix}:queue:${queueName}` : `queue:${queueName}`, prefix: config.prefix ? `${config.prefix}:queue:${queueName}` : `queue:${queueName}`,
}; };
} }
export function baseWorkerOptions(config: RedisOptions & RedisOptionsSource, queueName: typeof QUEUE[keyof typeof QUEUE]): Bull.WorkerOptions {
return {
connection: {
...config,
maxRetriesPerRequest: null,
keyPrefix: undefined,
},
prefix: config.prefix ? `${config.prefix}:queue:${queueName}` : `queue:${queueName}`,
skipLockRenewal: false,
lockDuration: 60 * 1000,
lockRenewTime: 30 * 1000,
stalledInterval: 90 * 1000,
};
}