diff options
Diffstat (limited to 'src/queue')
| -rw-r--r-- | src/queue/index.ts | 23 | ||||
| -rw-r--r-- | src/queue/initialize.ts | 18 | ||||
| -rw-r--r-- | src/queue/processors/deliver.ts | 19 | ||||
| -rw-r--r-- | src/queue/processors/inbox.ts | 1 | ||||
| -rw-r--r-- | src/queue/queues.ts | 7 |
5 files changed, 40 insertions, 28 deletions
diff --git a/src/queue/index.ts b/src/queue/index.ts index 163c57d691..9fb4595a35 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -1,4 +1,3 @@ -import * as Queue from 'bull'; import * as httpSignature from 'http-signature'; import config from '../config'; @@ -13,22 +12,7 @@ import { queueLogger } from './logger'; import { DriveFile } from '../models/entities/drive-file'; import { getJobInfo } from './get-job-info'; import { IActivity } from '../remote/activitypub/type'; - -function initializeQueue(name: string, limitPerSec = -1) { - return new Queue(name, { - redis: { - port: config.redis.port, - host: config.redis.host, - password: config.redis.pass, - db: config.redis.db || 0, - }, - prefix: config.redis.prefix ? `${config.redis.prefix}:queue` : 'queue', - limiter: limitPerSec > 0 ? { - max: limitPerSec * 5, - duration: 5000 - } : undefined - }); -} +import { dbQueue, deliverQueue, inboxQueue, objectStorageQueue } from './queues'; export type InboxJobData = { activity: IActivity, @@ -44,11 +28,6 @@ function renderError(e: Error): any { }; } -export const deliverQueue = initializeQueue('deliver', config.deliverJobPerSec || 128); -export const inboxQueue = initializeQueue('inbox', config.inboxJobPerSec || 16); -export const dbQueue = initializeQueue('db'); -export const objectStorageQueue = initializeQueue('objectStorage'); - const deliverLogger = queueLogger.createSubLogger('deliver'); const inboxLogger = queueLogger.createSubLogger('inbox'); const dbLogger = queueLogger.createSubLogger('db'); diff --git a/src/queue/initialize.ts b/src/queue/initialize.ts new file mode 100644 index 0000000000..92579531e4 --- /dev/null +++ b/src/queue/initialize.ts @@ -0,0 +1,18 @@ +import * as Queue from 'bull'; +import config from '../config'; + +export function initialize(name: string, limitPerSec = -1) { + return new Queue(name, { + redis: { + port: config.redis.port, + host: config.redis.host, + password: config.redis.pass, + db: config.redis.db || 0, + }, + prefix: config.redis.prefix ? `${config.redis.prefix}:queue` : 'queue', + limiter: limitPerSec > 0 ? { + max: limitPerSec * 5, + duration: 5000 + } : undefined + }); +} diff --git a/src/queue/processors/deliver.ts b/src/queue/processors/deliver.ts index cb7587ef81..a8b4ed4fe3 100644 --- a/src/queue/processors/deliver.ts +++ b/src/queue/processors/deliver.ts @@ -7,11 +7,15 @@ import { instanceChart } from '../../services/chart'; import { fetchInstanceMetadata } from '../../services/fetch-instance-metadata'; import { fetchMeta } from '../../misc/fetch-meta'; import { toPuny } from '../../misc/convert-host'; +import { Cache } from '../../misc/cache'; +import { Instance } from '../../models/entities/instance'; const logger = new Logger('deliver'); let latest: string | null = null; +const suspendedHostsCache = new Cache<Instance[]>(1000 * 60 * 60); + export default async (job: Bull.Job) => { const { host } = new URL(job.data.to); @@ -22,12 +26,15 @@ export default async (job: Bull.Job) => { } // isSuspendedなら中断 - const suspendedHosts = await Instances.find({ - where: { - isSuspended: true - }, - cache: 60 * 1000 - }); + let suspendedHosts = suspendedHostsCache.get(null); + if (suspendedHosts == null) { + suspendedHosts = await Instances.find({ + where: { + isSuspended: true + }, + }); + suspendedHostsCache.set(null, suspendedHosts); + } if (suspendedHosts.map(x => x.host).includes(toPuny(host))) { return 'skip (suspended)'; } diff --git a/src/queue/processors/inbox.ts b/src/queue/processors/inbox.ts index b4e8b85a46..a5822ff25f 100644 --- a/src/queue/processors/inbox.ts +++ b/src/queue/processors/inbox.ts @@ -40,6 +40,7 @@ export default async (job: Bull.Job<InboxJobData>): Promise<string> => { return `Old keyId is no longer supported. ${keyIdLower}`; } + // TDOO: キャッシュ const dbResolver = new DbResolver(); // HTTP-Signature keyIdを元にDBから取得 diff --git a/src/queue/queues.ts b/src/queue/queues.ts new file mode 100644 index 0000000000..d589d9f7da --- /dev/null +++ b/src/queue/queues.ts @@ -0,0 +1,7 @@ +import config from '../config'; +import { initialize as initializeQueue } from './initialize'; + +export const deliverQueue = initializeQueue('deliver', config.deliverJobPerSec || 128); +export const inboxQueue = initializeQueue('inbox', config.inboxJobPerSec || 16); +export const dbQueue = initializeQueue('db'); +export const objectStorageQueue = initializeQueue('objectStorage'); |