summaryrefslogtreecommitdiff
path: root/src/queue
diff options
context:
space:
mode:
authorsyuilo <Syuilotan@yahoo.co.jp>2021-03-22 15:27:08 +0900
committersyuilo <Syuilotan@yahoo.co.jp>2021-03-22 15:27:08 +0900
commit52d577c7dd7bf87b3fae34f539bb6e656c7c0ed2 (patch)
tree9805c625a7fba9d8631db8a92772b2772d8632ec /src/queue
parentMerge branch 'develop' (diff)
parent12.75.0 (diff)
downloadmisskey-52d577c7dd7bf87b3fae34f539bb6e656c7c0ed2.tar.gz
misskey-52d577c7dd7bf87b3fae34f539bb6e656c7c0ed2.tar.bz2
misskey-52d577c7dd7bf87b3fae34f539bb6e656c7c0ed2.zip
Merge branch 'develop'
Diffstat (limited to 'src/queue')
-rw-r--r--src/queue/index.ts23
-rw-r--r--src/queue/initialize.ts18
-rw-r--r--src/queue/processors/deliver.ts19
-rw-r--r--src/queue/processors/inbox.ts1
-rw-r--r--src/queue/queues.ts7
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');