Merge branch 'support-mcaptcha' of ssh://github.com/chocolate-pie/mis…

…skey into support-mcaptcha
chocolate-pie · Jan 6, 2024 · e0afa5c · e0afa5c
2 parents cda808d + 727e122
commit e0afa5c
Show file tree

Hide file tree

Showing 10 changed files with 39 additions and 16 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -22,6 +22,10 @@
 - Fix: v2023.12.0で追加された「モデレーターがユーザーのアイコンもしくはバナー画像を未設定状態にできる機能」が管理画面上で正しく表示されていない問題を修正
 - Enhance: チャンネルノートのピン留めをノートのメニューからできるよ
 
+### Server
+- Enhance: 連合先のレートリミットに引っかかった際にリトライするようになりました
+- Enhance: ActivityPub Deliver queueでBodyを事前処理するように (#12916)
+
 ## 2023.12.2
 
 ### General

diff --git a/packages/backend/src/core/QueueService.ts b/packages/backend/src/core/QueueService.ts
@@ -16,6 +16,7 @@ import type { DbQueue, DeliverQueue, EndedPollNotificationQueue, InboxQueue, Obj
 import type { DbJobData, DeliverJobData, RelationshipJobData, ThinUser } from '../queue/types.js';
 import type httpSignature from '@peertube/http-signature';
 import type * as Bull from 'bullmq';
+import { ApRequestCreator } from '@/core/activitypub/ApRequestService.js';
 
 @Injectable()
 export class QueueService {
@@ -74,11 +75,15 @@ export class QueueService {
 		if (content == null) return null;
 		if (to == null) return null;
 
+		const contentBody = JSON.stringify(content);
+		const digest = ApRequestCreator.createDigest(contentBody);
+
 		const data: DeliverJobData = {
 			user: {
 				id: user.id,
 			},
-			content,
+			content: contentBody,
+			digest,
 			to,
 			isSharedInbox,
 		};
@@ -103,6 +108,8 @@ export class QueueService {
 	@bindThis
 	public async deliverMany(user: ThinUser, content: IActivity | null, inboxes: Map<string, boolean>) {
 		if (content == null) return null;
+		const contentBody = JSON.stringify(content);
+		const digest = ApRequestCreator.createDigest(contentBody);
 
 		const opts = {
 			attempts: this.config.deliverJobMaxAttempts ?? 12,
@@ -117,7 +124,8 @@ export class QueueService {
 			name: d[0],
 			data: {
 				user,
-				content,
+				content: contentBody,
+				digest,
 				to: d[0],
 				isSharedInbox: d[1],
 			} as DeliverJobData,

diff --git a/packages/backend/src/core/activitypub/ApInboxService.ts b/packages/backend/src/core/activitypub/ApInboxService.ts
@@ -97,6 +97,8 @@ export class ApInboxService {
 				} catch (err) {
 					if (err instanceof Error || typeof err === 'string') {
 						this.logger.error(err);
+					} else {
+						throw err;
 					}
 				}
 			}
@@ -256,7 +258,7 @@ export class ApInboxService {
 
 		const targetUri = getApId(activity.object);
 
-		this.announceNote(actor, activity, targetUri);
+		await this.announceNote(actor, activity, targetUri);
 	}
 
 	@bindThis
@@ -288,7 +290,7 @@ export class ApInboxService {
 			} catch (err) {
 				// 対象が4xxならスキップ
 				if (err instanceof StatusError) {
-					if (err.isClientError) {
+					if (!err.isRetryable) {
 						this.logger.warn(`Ignored announce target ${targetUri} - ${err.statusCode}`);
 						return;
 					}
@@ -373,7 +375,7 @@ export class ApInboxService {
 		});
 
 		if (isPost(object)) {
-			this.createNote(resolver, actor, object, false, activity);
+			await this.createNote(resolver, actor, object, false, activity);
 		} else {
 			this.logger.warn(`Unknown type: ${getApType(object)}`);
 		}
@@ -404,7 +406,7 @@ export class ApInboxService {
 			await this.apNoteService.createNote(note, resolver, silent);
 			return 'ok';
 		} catch (err) {
-			if (err instanceof StatusError && err.isClientError) {
+			if (err instanceof StatusError && !err.isRetryable) {
 				return `skip ${err.statusCode}`;
 			} else {
 				throw err;

diff --git a/packages/backend/src/core/activitypub/ApRequestService.ts b/packages/backend/src/core/activitypub/ApRequestService.ts
@@ -34,9 +34,9 @@ type PrivateKey = {
 };
 
 export class ApRequestCreator {
-	static createSignedPost(args: { key: PrivateKey, url: string, body: string, additionalHeaders: Record<string, string> }): Signed {
+	static createSignedPost(args: { key: PrivateKey, url: string, body: string, digest?: string, additionalHeaders: Record<string, string> }): Signed {
 		const u = new URL(args.url);
-		const digestHeader = `SHA-256=${crypto.createHash('sha256').update(args.body).digest('base64')}`;
+		const digestHeader = args.digest ?? this.createDigest(args.body);
 
 		const request: Request = {
 			url: u.href,
@@ -59,6 +59,10 @@ export class ApRequestCreator {
 		};
 	}
 
+	static createDigest(body: string) {
+		return `SHA-256=${crypto.createHash('sha256').update(body).digest('base64')}`;
+	}
+
 	static createSignedGet(args: { key: PrivateKey, url: string, additionalHeaders: Record<string, string> }): Signed {
 		const u = new URL(args.url);
 
@@ -145,8 +149,8 @@ export class ApRequestService {
 	}
 
 	@bindThis
-	public async signedPost(user: { id: MiUser['id'] }, url: string, object: unknown): Promise<void> {
-		const body = JSON.stringify(object);
+	public async signedPost(user: { id: MiUser['id'] }, url: string, object: unknown, digest?: string): Promise<void> {
+		const body = typeof object === 'string' ? object : JSON.stringify(object);
 
 		const keypair = await this.userKeypairService.getUserKeypair(user.id);
 
@@ -157,6 +161,7 @@ export class ApRequestService {
 			},
 			url,
 			body,
+			digest,
 			additionalHeaders: {
 			},
 		});

diff --git a/packages/backend/src/core/activitypub/models/ApNoteService.ts b/packages/backend/src/core/activitypub/models/ApNoteService.ts
@@ -216,7 +216,7 @@ export class ApNoteService {
 					return { status: 'ok', res };
 				} catch (e) {
 					return {
-						status: (e instanceof StatusError && e.isClientError) ? 'permerror' : 'temperror',
+						status: (e instanceof StatusError && !e.isRetryable) ? 'permerror' : 'temperror',
 					};
 				}
 			};

diff --git a/packages/backend/src/misc/status-error.ts b/packages/backend/src/misc/status-error.ts
@@ -7,12 +7,14 @@ export class StatusError extends Error {
 	public statusCode: number;
 	public statusMessage?: string;
 	public isClientError: boolean;
+	public isRetryable: boolean;
 
 	constructor(message: string, statusCode: number, statusMessage?: string) {
 		super(message);
 		this.name = 'StatusError';
 		this.statusCode = statusCode;
 		this.statusMessage = statusMessage;
 		this.isClientError = typeof this.statusCode === 'number' && this.statusCode >= 400 && this.statusCode < 500;
+		this.isRetryable = !this.isClientError || this.statusCode === 429;
 	}
 }
diff --git a/packages/backend/src/queue/processors/DeliverProcessorService.ts b/packages/backend/src/queue/processors/DeliverProcessorService.ts
@@ -72,7 +72,7 @@ export class DeliverProcessorService {
 		}
 
 		try {
-			await this.apRequestService.signedPost(job.data.user, job.data.to, job.data.content);
+			await this.apRequestService.signedPost(job.data.user, job.data.to, job.data.content, job.data.digest);
 
 			// Update stats
 			this.federatedInstanceService.fetch(host).then(i => {
@@ -111,7 +111,7 @@ export class DeliverProcessorService {
 
 			if (res instanceof StatusError) {
 				// 4xx
-				if (res.isClientError) {
+				if (!res.isRetryable) {
 					// 相手が閉鎖していることを明示しているため、配送停止する
 					if (job.data.isSharedInbox && res.statusCode === 410) {
 						this.federatedInstanceService.fetch(host).then(i => {

diff --git a/packages/backend/src/queue/processors/InboxProcessorService.ts b/packages/backend/src/queue/processors/InboxProcessorService.ts
@@ -85,7 +85,7 @@ export class InboxProcessorService {
 			} catch (err) {
 				// 対象が4xxならスキップ
 				if (err instanceof StatusError) {
-					if (err.isClientError) {
+					if (!err.isRetryable) {
 						throw new Bull.UnrecoverableError(`skip: Ignored deleted actors on both ends ${activity.actor} - ${err.statusCode}`);
 					}
 					throw new Error(`Error in actor ${activity.actor} - ${err.statusCode}`);

diff --git a/packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts b/packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts
@@ -71,7 +71,7 @@ export class WebhookDeliverProcessorService {
 
 			if (res instanceof StatusError) {
 				// 4xx
-				if (res.isClientError) {
+				if (!res.isRetryable) {
 					throw new Bull.UnrecoverableError(`${res.statusCode} ${res.statusMessage}`);
 				}
 

diff --git a/packages/backend/src/queue/types.ts b/packages/backend/src/queue/types.ts
@@ -15,7 +15,9 @@ export type DeliverJobData = {
 	/** Actor */
 	user: ThinUser;
 	/** Activity */
-	content: unknown;
+	content: string;
+	/** Digest header */
+	digest: string;
 	/** inbox URL to deliver */
 	to: string;
 	/** whether it is sharedInbox */