From e7643da3bd2f7c7fdb8d9976a0bc164688e2d7df Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E3=81=BE=E3=81=A3=E3=81=A1=E3=82=83=E3=81=A8=E3=83=BC?=
 =?UTF-8?q?=E3=81=AB=E3=82=85?=
 <17376330+u1-liquid@users.noreply.github.com>
Date: Mon, 1 Jan 2024 02:57:23 +0900
Subject: [PATCH] =?UTF-8?q?spec(Queue):=20lockDuration=E3=81=A8stalledInte?=
 =?UTF-8?q?rval=E3=82=92=E5=A2=97=E3=82=84=E3=81=99=20(MisskeyIO#308)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../backend/src/queue/QueueProcessorService.ts | 18 +++++++++---------
 packages/backend/src/queue/const.ts            | 15 +++++++++++++++
 2 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts
index 9df372b3e..1b9b39808 100644
--- a/packages/backend/src/queue/QueueProcessorService.ts
+++ b/packages/backend/src/queue/QueueProcessorService.ts
@@ -40,7 +40,7 @@ import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMu
 import { CleanProcessorService } from './processors/CleanProcessorService.js';
 import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.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
 function httpRelatedBackoff(attemptsMade: number) {
@@ -146,7 +146,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 				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,
 		});
 
@@ -185,7 +185,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 				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,
 		});
 
@@ -201,7 +201,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 
 		//#region deliver
 		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,
 			concurrency: this.config.deliverJobConcurrency ?? 128,
 			limiter: {
@@ -225,7 +225,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 
 		//#region inbox
 		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,
 			concurrency: this.config.inboxJobConcurrency ?? 16,
 			limiter: {
@@ -249,7 +249,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 
 		//#region webhook deliver
 		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,
 			concurrency: 64,
 			limiter: {
@@ -281,7 +281,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 				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,
 			concurrency: this.config.relashionshipJobConcurrency ?? 16,
 			limiter: {
@@ -308,7 +308,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 				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,
 			concurrency: 16,
 		});
@@ -325,7 +325,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 
 		//#region ended poll notification
 		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,
 		});
 		//#endregion
diff --git a/packages/backend/src/queue/const.ts b/packages/backend/src/queue/const.ts
index 55869095c..d0ce23c05 100644
--- a/packages/backend/src/queue/const.ts
+++ b/packages/backend/src/queue/const.ts
@@ -28,3 +28,18 @@ export function baseQueueOptions(config: RedisOptions & RedisOptionsSource, queu
 		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,
+	};
+}