From dea1188714946a8d3d0ed7dfa3b385a623e24c2a Mon Sep 17 00:00:00 2001
From: MeiMei <30769358+mei23@users.noreply.github.com>
Date: Mon, 20 Jan 2020 14:14:09 +0900
Subject: [PATCH] =?UTF-8?q?AP=E3=81=AE=E6=B5=81=E9=87=8F=E5=88=B6=E9=99=90?=
 =?UTF-8?q?=E3=81=A8=E3=83=AA=E3=83=88=E3=83=A9=E3=82=A4=E6=9C=9F=E9=96=93?=
 =?UTF-8?q?=E3=81=AE=E5=A4=89=E6=9B=B4=20(#5734)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* AP rate limit

* AP Job attempts

* fix
---
 .config/example.yml |  8 ++++++++
 src/config/types.ts |  4 ++++
 src/queue/index.ts  | 16 ++++++++++------
 3 files changed, 22 insertions(+), 6 deletions(-)

diff --git a/.config/example.yml b/.config/example.yml
index 7afa56fbe..cd08f76d6 100644
--- a/.config/example.yml
+++ b/.config/example.yml
@@ -125,6 +125,14 @@ autoAdmin: true
 # deliverJobConcurrency: 128
 # inboxJobConcurrency: 16
 
+# Job rate limiter
+# deliverJobPerSec: 128
+# inboxJobPerSec: 16
+
+# Job attempts
+# deliverJobMaxAttempts: 12
+# inboxJobMaxAttempts: 8
+
 # IP address family used for outgoing request (ipv4, ipv6 or dual)
 #outgoingAddressFamily: ipv4
 
diff --git a/src/config/types.ts b/src/config/types.ts
index 2bf94af74..aeb2c1233 100644
--- a/src/config/types.ts
+++ b/src/config/types.ts
@@ -47,6 +47,10 @@ export type Source = {
 
 	deliverJobConcurrency?: number;
 	inboxJobConcurrency?: number;
+	deliverJobPerSec?: number;
+	inboxJobPerSec?: number;
+	deliverJobMaxAttempts?: number;
+	inboxJobMaxAttempts?: number;
 
 	syslog: {
 		host: string;
diff --git a/src/queue/index.ts b/src/queue/index.ts
index ad7453353..b1437da8b 100644
--- a/src/queue/index.ts
+++ b/src/queue/index.ts
@@ -13,7 +13,7 @@ import { queueLogger } from './logger';
 import { DriveFile } from '../models/entities/drive-file';
 import { getJobInfo } from './get-job-info';
 
-function initializeQueue(name: string) {
+function initializeQueue(name: string, limitPerSec = -1) {
 	return new Queue(name, {
 		redis: {
 			port: config.redis.port,
@@ -21,7 +21,11 @@ function initializeQueue(name: string) {
 			password: config.redis.pass,
 			db: config.redis.db || 0,
 		},
-		prefix: config.redis.prefix ? `${config.redis.prefix}:queue` : 'queue'
+		prefix: config.redis.prefix ? `${config.redis.prefix}:queue` : 'queue',
+		limiter: limitPerSec > 0 ? {
+			max: limitPerSec * 5,
+			duration: 5000
+		} : undefined
 	});
 }
 
@@ -33,8 +37,8 @@ function renderError(e: Error): any {
 	};
 }
 
-export const deliverQueue = initializeQueue('deliver');
-export const inboxQueue = initializeQueue('inbox');
+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');
 
@@ -85,7 +89,7 @@ export function deliver(user: ILocalUser, content: any, to: any) {
 	};
 
 	return deliverQueue.add(data, {
-		attempts: 8,
+		attempts: config.deliverJobMaxAttempts || 12,
 		backoff: {
 			type: 'exponential',
 			delay: 60 * 1000
@@ -102,7 +106,7 @@ export function inbox(activity: any, signature: httpSignature.IParsedSignature)
 	};
 
 	return inboxQueue.add(data, {
-		attempts: 8,
+		attempts: config.inboxJobMaxAttempts || 8,
 		backoff: {
 			type: 'exponential',
 			delay: 1000