oscar-surf/misskey
oscar-surf/misskey
0
0
Fork 0
Code Issues Activity

Merge upstream

Browse source
This commit is contained in:
ASTRO:? 2024-12-30 23:28:39 +09:00
commit 5f931855be
No known key found for this signature in database
GPG key ID: 8947F3AF5B0B4BFE
147 changed files with 1778 additions and 1044 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -45,7 +45,7 @@ import { CleanProcessorService } from './processors/CleanProcessorService.js';
import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js';
import { AutoNoteRemovalProcessorService } from './processors/AutoNoteRemovalProcessorService.js';
import { QueueLoggerService } from './QueueLoggerService.js';
import { QUEUE, baseWorkerOptions } from './const.js';
import { QUEUE, baseWorkerOptions, formatQueueName } from './const.js';
// ref. https://github.com/misskey-dev/misskey/pull/7635#issue-971097019
function httpRelatedBackoff(attemptsMade: number) {
@ -217,7 +217,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
//#region deliver
this.deliverQueueWorkers = this.config.redisForDeliverQueues
.filter((_, index) => process.env.QUEUE_WORKER_INDEX == null || index === Number.parseInt(process.env.QUEUE_WORKER_INDEX, 10))
.map(config => new Bull.Worker(QUEUE.DELIVER, (job) => this.deliverProcessorService.process(job), {
.map(config => new Bull.Worker(formatQueueName(config, QUEUE.DELIVER), (job) => this.deliverProcessorService.process(job), {
...baseWorkerOptions(config, this.config.bullmqWorkerOptions, QUEUE.DELIVER),
autorun: false,
concurrency: this.config.deliverJobConcurrency ?? 128,
@ -245,7 +245,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
//#region inbox
this.inboxQueueWorkers = this.config.redisForInboxQueues
.filter((_, index) => process.env.QUEUE_WORKER_INDEX == null || index === Number.parseInt(process.env.QUEUE_WORKER_INDEX, 10))
.map(config => new Bull.Worker(QUEUE.INBOX, (job) => this.inboxProcessorService.process(job), {
.map(config => new Bull.Worker(formatQueueName(config, QUEUE.INBOX), (job) => this.inboxProcessorService.process(job), {
...baseWorkerOptions(config, this.config.bullmqWorkerOptions, QUEUE.INBOX),
autorun: false,
concurrency: this.config.inboxJobConcurrency ?? 16,
@ -297,7 +297,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
//#region relationship
this.relationshipQueueWorkers = this.config.redisForRelationshipQueues
.filter((_, index) => process.env.QUEUE_WORKER_INDEX == null || index === Number.parseInt(process.env.QUEUE_WORKER_INDEX, 10))
.map(config => new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
.map(config => new Bull.Worker(formatQueueName(config, QUEUE.RELATIONSHIP), (job) => {
switch (job.name) {
case 'follow': return this.relationshipProcessorService.processFollow(job);
case 'unfollow': return this.relationshipProcessorService.processUnfollow(job);