enhance: ActivityPub 周りで 429 を受け取った際にリトライするように (MisskeyIO#324)

packages/backend/src/core/activitypub/ApInboxService.ts

@@ -288,7 +288,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;
 					}
@@ -404,7 +404,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;

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',
 					};
 				}
 			};

packages/backend/src/misc/status-error.ts

@@ -7,6 +7,7 @@ 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);
@@ -14,5 +15,6 @@ export class StatusError extends Error {
 		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;
 	}
 }

packages/backend/src/queue/processors/DeliverProcessorService.ts

@@ -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 => {

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}`);

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}`);
 				}