Merge pull request #14233 from misskey-dev/develop

Release: 2024.7.0

8 files changed, 459 insertions, 174 deletions

diff --git a/packages/backend/src/queue/QueueProcessorModule.ts b/packages/backend/src/queue/QueueProcessorModule.ts
index 8086158997..a1fd38fcc5 100644
--- a/packages/backend/src/queue/QueueProcessorModule.ts
+++ b/packages/backend/src/queue/QueueProcessorModule.ts
@@ -11,7 +11,8 @@ import { QueueProcessorService } from './QueueProcessorService.js';
 import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
 import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
 import { InboxProcessorService } from './processors/InboxProcessorService.js';
-import { WebhookDeliverProcessorService } from './processors/WebhookDeliverProcessorService.js';
+import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js';
+import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js';
 import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMutingsProcessorService.js';
 import { CleanChartsProcessorService } from './processors/CleanChartsProcessorService.js';
 import { CleanProcessorService } from './processors/CleanProcessorService.js';
@@ -71,7 +72,8 @@ import { RelationshipProcessorService } from './processors/RelationshipProcessor
 		DeleteFileProcessorService,
 		CleanRemoteFilesProcessorService,
 		RelationshipProcessorService,
-		WebhookDeliverProcessorService,
+		UserWebhookDeliverProcessorService,
+		SystemWebhookDeliverProcessorService,
 		EndedPollNotificationProcessorService,
 		DeliverProcessorService,
 		InboxProcessorService,
diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts
index ce999d9cef..7bd74f3210 100644
--- a/packages/backend/src/queue/QueueProcessorService.ts
+++ b/packages/backend/src/queue/QueueProcessorService.ts
@@ -5,11 +5,13 @@
 
 import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common';
 import * as Bull from 'bullmq';
+import * as Sentry from '@sentry/node';
 import type { Config } from '@/config.js';
 import { DI } from '@/di-symbols.js';
 import type Logger from '@/logger.js';
 import { bindThis } from '@/decorators.js';
-import { WebhookDeliverProcessorService } from './processors/WebhookDeliverProcessorService.js';
+import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js';
+import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js';
 import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
 import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
 import { InboxProcessorService } from './processors/InboxProcessorService.js';
@@ -75,7 +77,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
 	private dbQueueWorker: Bull.Worker;
 	private deliverQueueWorker: Bull.Worker;
 	private inboxQueueWorker: Bull.Worker;
-	private webhookDeliverQueueWorker: Bull.Worker;
+	private userWebhookDeliverQueueWorker: Bull.Worker;
+	private systemWebhookDeliverQueueWorker: Bull.Worker;
 	private relationshipQueueWorker: Bull.Worker;
 	private objectStorageQueueWorker: Bull.Worker;
 	private endedPollNotificationQueueWorker: Bull.Worker;
@@ -85,7 +88,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		private config: Config,
 
 		private queueLoggerService: QueueLoggerService,
-		private webhookDeliverProcessorService: WebhookDeliverProcessorService,
+		private userWebhookDeliverProcessorService: UserWebhookDeliverProcessorService,
+		private systemWebhookDeliverProcessorService: SystemWebhookDeliverProcessorService,
 		private endedPollNotificationProcessorService: EndedPollNotificationProcessorService,
 		private deliverProcessorService: DeliverProcessorService,
 		private inboxProcessorService: InboxProcessorService,
@@ -135,199 +139,367 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		}
 
 		//#region system
-		this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => {
-			switch (job.name) {
-				case 'tickCharts': return this.tickChartsProcessorService.process();
-				case 'resyncCharts': return this.resyncChartsProcessorService.process();
-				case 'cleanCharts': return this.cleanChartsProcessorService.process();
-				case 'aggregateRetention': return this.aggregateRetentionProcessorService.process();
-				case 'checkExpiredMutings': return this.checkExpiredMutingsProcessorService.process();
-				case 'clean': return this.cleanProcessorService.process();
-				default: throw new Error(`unrecognized job type ${job.name} for system`);
-			}
-		}, {
-			...baseQueueOptions(this.config, QUEUE.SYSTEM),
-			autorun: false,
-		});
+		{
+			const processer = (job: Bull.Job) => {
+				switch (job.name) {
+					case 'tickCharts': return this.tickChartsProcessorService.process();
+					case 'resyncCharts': return this.resyncChartsProcessorService.process();
+					case 'cleanCharts': return this.cleanChartsProcessorService.process();
+					case 'aggregateRetention': return this.aggregateRetentionProcessorService.process();
+					case 'checkExpiredMutings': return this.checkExpiredMutingsProcessorService.process();
+					case 'clean': return this.cleanProcessorService.process();
+					default: throw new Error(`unrecognized job type ${job.name} for system`);
+				}
+			};
+
+			this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: System: ' + job.name }, () => processer(job));
+				} else {
+					return processer(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.SYSTEM),
+				autorun: false,
+			});
 
-		const systemLogger = this.logger.createSubLogger('system');
+			const logger = this.logger.createSubLogger('system');
 
-		this.systemQueueWorker
-			.on('active', (job) => systemLogger.debug(`active id=${job.id}`))
-			.on('completed', (job, result) => systemLogger.debug(`completed(${result}) id=${job.id}`))
-			.on('failed', (job, err) => systemLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
-			.on('error', (err: Error) => systemLogger.error(`error ${err.stack}`, { e: renderError(err) }))
-			.on('stalled', (jobId) => systemLogger.warn(`stalled id=${jobId}`));
+			this.systemQueueWorker
+				.on('active', (job) => logger.debug(`active id=${job.id}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+				.on('failed', (job, err: Error) => {
+					logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: System: ${job?.name ?? '?'}: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
 		//#endregion
 
 		//#region db
-		this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
-			switch (job.name) {
-				case 'deleteDriveFiles': return this.deleteDriveFilesProcessorService.process(job);
-				case 'exportCustomEmojis': return this.exportCustomEmojisProcessorService.process(job);
-				case 'exportNotes': return this.exportNotesProcessorService.process(job);
-				case 'exportClips': return this.exportClipsProcessorService.process(job);
-				case 'exportFavorites': return this.exportFavoritesProcessorService.process(job);
-				case 'exportFollowing': return this.exportFollowingProcessorService.process(job);
-				case 'exportMuting': return this.exportMutingProcessorService.process(job);
-				case 'exportBlocking': return this.exportBlockingProcessorService.process(job);
-				case 'exportUserLists': return this.exportUserListsProcessorService.process(job);
-				case 'exportAntennas': return this.exportAntennasProcessorService.process(job);
-				case 'importFollowing': return this.importFollowingProcessorService.process(job);
-				case 'importFollowingToDb': return this.importFollowingProcessorService.processDb(job);
-				case 'importMuting': return this.importMutingProcessorService.process(job);
-				case 'importBlocking': return this.importBlockingProcessorService.process(job);
-				case 'importBlockingToDb': return this.importBlockingProcessorService.processDb(job);
-				case 'importUserLists': return this.importUserListsProcessorService.process(job);
-				case 'importCustomEmojis': return this.importCustomEmojisProcessorService.process(job);
-				case 'importAntennas': return this.importAntennasProcessorService.process(job);
-				case 'deleteAccount': return this.deleteAccountProcessorService.process(job);
-				default: throw new Error(`unrecognized job type ${job.name} for db`);
-			}
-		}, {
-			...baseQueueOptions(this.config, QUEUE.DB),
-			autorun: false,
-		});
+		{
+			const processer = (job: Bull.Job) => {
+				switch (job.name) {
+					case 'deleteDriveFiles': return this.deleteDriveFilesProcessorService.process(job);
+					case 'exportCustomEmojis': return this.exportCustomEmojisProcessorService.process(job);
+					case 'exportNotes': return this.exportNotesProcessorService.process(job);
+					case 'exportClips': return this.exportClipsProcessorService.process(job);
+					case 'exportFavorites': return this.exportFavoritesProcessorService.process(job);
+					case 'exportFollowing': return this.exportFollowingProcessorService.process(job);
+					case 'exportMuting': return this.exportMutingProcessorService.process(job);
+					case 'exportBlocking': return this.exportBlockingProcessorService.process(job);
+					case 'exportUserLists': return this.exportUserListsProcessorService.process(job);
+					case 'exportAntennas': return this.exportAntennasProcessorService.process(job);
+					case 'importFollowing': return this.importFollowingProcessorService.process(job);
+					case 'importFollowingToDb': return this.importFollowingProcessorService.processDb(job);
+					case 'importMuting': return this.importMutingProcessorService.process(job);
+					case 'importBlocking': return this.importBlockingProcessorService.process(job);
+					case 'importBlockingToDb': return this.importBlockingProcessorService.processDb(job);
+					case 'importUserLists': return this.importUserListsProcessorService.process(job);
+					case 'importCustomEmojis': return this.importCustomEmojisProcessorService.process(job);
+					case 'importAntennas': return this.importAntennasProcessorService.process(job);
+					case 'deleteAccount': return this.deleteAccountProcessorService.process(job);
+					default: throw new Error(`unrecognized job type ${job.name} for db`);
+				}
+			};
+
+			this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: DB: ' + job.name }, () => processer(job));
+				} else {
+					return processer(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.DB),
+				autorun: false,
+			});
 
-		const dbLogger = this.logger.createSubLogger('db');
+			const logger = this.logger.createSubLogger('db');
 
-		this.dbQueueWorker
-			.on('active', (job) => dbLogger.debug(`active id=${job.id}`))
-			.on('completed', (job, result) => dbLogger.debug(`completed(${result}) id=${job.id}`))
-			.on('failed', (job, err) => dbLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
-			.on('error', (err: Error) => dbLogger.error(`error ${err.stack}`, { e: renderError(err) }))
-			.on('stalled', (jobId) => dbLogger.warn(`stalled id=${jobId}`));
+			this.dbQueueWorker
+				.on('active', (job) => logger.debug(`active id=${job.id}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+				.on('failed', (job, err) => {
+					logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: DB: ${job?.name ?? '?'}: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
 		//#endregion
 
 		//#region deliver
-		this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => this.deliverProcessorService.process(job), {
-			...baseQueueOptions(this.config, QUEUE.DELIVER),
-			autorun: false,
-			concurrency: this.config.deliverJobConcurrency ?? 128,
-			limiter: {
-				max: this.config.deliverJobPerSec ?? 128,
-				duration: 1000,
-			},
-			settings: {
-				backoffStrategy: httpRelatedBackoff,
-			},
-		});
+		{
+			this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: Deliver' }, () => this.deliverProcessorService.process(job));
+				} else {
+					return this.deliverProcessorService.process(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.DELIVER),
+				autorun: false,
+				concurrency: this.config.deliverJobConcurrency ?? 128,
+				limiter: {
+					max: this.config.deliverJobPerSec ?? 128,
+					duration: 1000,
+				},
+				settings: {
+					backoffStrategy: httpRelatedBackoff,
+				},
+			});
 
-		const deliverLogger = this.logger.createSubLogger('deliver');
+			const logger = this.logger.createSubLogger('deliver');
 
-		this.deliverQueueWorker
-			.on('active', (job) => deliverLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
-			.on('completed', (job, result) => deliverLogger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
-			.on('failed', (job, err) => deliverLogger.warn(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`))
-			.on('error', (err: Error) => deliverLogger.error(`error ${err.stack}`, { e: renderError(err) }))
-			.on('stalled', (jobId) => deliverLogger.warn(`stalled id=${jobId}`));
+			this.deliverQueueWorker
+				.on('active', (job) => logger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
+				.on('failed', (job, err) => {
+					logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: Deliver: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
 		//#endregion
 
 		//#region inbox
-		this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => this.inboxProcessorService.process(job), {
-			...baseQueueOptions(this.config, QUEUE.INBOX),
-			autorun: false,
-			concurrency: this.config.inboxJobConcurrency ?? 16,
-			limiter: {
-				max: this.config.inboxJobPerSec ?? 32,
-				duration: 1000,
-			},
-			settings: {
-				backoffStrategy: httpRelatedBackoff,
-			},
-		});
+		{
+			this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: Inbox' }, () => this.inboxProcessorService.process(job));
+				} else {
+					return this.inboxProcessorService.process(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.INBOX),
+				autorun: false,
+				concurrency: this.config.inboxJobConcurrency ?? 16,
+				limiter: {
+					max: this.config.inboxJobPerSec ?? 32,
+					duration: 1000,
+				},
+				settings: {
+					backoffStrategy: httpRelatedBackoff,
+				},
+			});
 
-		const inboxLogger = this.logger.createSubLogger('inbox');
+			const logger = this.logger.createSubLogger('inbox');
 
-		this.inboxQueueWorker
-			.on('active', (job) => inboxLogger.debug(`active ${getJobInfo(job, true)}`))
-			.on('completed', (job, result) => inboxLogger.debug(`completed(${result}) ${getJobInfo(job, true)}`))
-			.on('failed', (job, err) => inboxLogger.warn(`failed(${err.stack}) ${getJobInfo(job)} activity=${job ? (job.data.activity ? job.data.activity.id : 'none') : '-'}`, { job, e: renderError(err) }))
-			.on('error', (err: Error) => inboxLogger.error(`error ${err.stack}`, { e: renderError(err) }))
-			.on('stalled', (jobId) => inboxLogger.warn(`stalled id=${jobId}`));
+			this.inboxQueueWorker
+				.on('active', (job) => logger.debug(`active ${getJobInfo(job, true)}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)}`))
+				.on('failed', (job, err) => {
+					logger.error(`failed(${err.stack}) ${getJobInfo(job)} activity=${job ? (job.data.activity ? job.data.activity.id : 'none') : '-'}`, { job, e: renderError(err) });
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: Inbox: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
 		//#endregion
 
-		//#region webhook deliver
-		this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => this.webhookDeliverProcessorService.process(job), {
-			...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER),
-			autorun: false,
-			concurrency: 64,
-			limiter: {
-				max: 64,
-				duration: 1000,
-			},
-			settings: {
-				backoffStrategy: httpRelatedBackoff,
-			},
-		});
+		//#region user-webhook deliver
+		{
+			this.userWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.USER_WEBHOOK_DELIVER, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: UserWebhookDeliver' }, () => this.userWebhookDeliverProcessorService.process(job));
+				} else {
+					return this.userWebhookDeliverProcessorService.process(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER),
+				autorun: false,
+				concurrency: 64,
+				limiter: {
+					max: 64,
+					duration: 1000,
+				},
+				settings: {
+					backoffStrategy: httpRelatedBackoff,
+				},
+			});
 
-		const webhookLogger = this.logger.createSubLogger('webhook');
+			const logger = this.logger.createSubLogger('user-webhook');
 
-		this.webhookDeliverQueueWorker
-			.on('active', (job) => webhookLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
-			.on('completed', (job, result) => webhookLogger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
-			.on('failed', (job, err) => webhookLogger.warn(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`))
-			.on('error', (err: Error) => webhookLogger.error(`error ${err.stack}`, { e: renderError(err) }))
-			.on('stalled', (jobId) => webhookLogger.warn(`stalled id=${jobId}`));
+			this.userWebhookDeliverQueueWorker
+				.on('active', (job) => logger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
+				.on('failed', (job, err) => {
+					logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: UserWebhookDeliver: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
+		//#endregion
+
+		//#region system-webhook deliver
+		{
+			this.systemWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.SYSTEM_WEBHOOK_DELIVER, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: SystemWebhookDeliver' }, () => this.systemWebhookDeliverProcessorService.process(job));
+				} else {
+					return this.systemWebhookDeliverProcessorService.process(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.SYSTEM_WEBHOOK_DELIVER),
+				autorun: false,
+				concurrency: 16,
+				limiter: {
+					max: 16,
+					duration: 1000,
+				},
+				settings: {
+					backoffStrategy: httpRelatedBackoff,
+				},
+			});
+
+			const logger = this.logger.createSubLogger('system-webhook');
+
+			this.systemWebhookDeliverQueueWorker
+				.on('active', (job) => logger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
+				.on('failed', (job, err) => {
+					logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: SystemWebhookDeliver: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
 		//#endregion
 
 		//#region relationship
-		this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
-			switch (job.name) {
-				case 'follow': return this.relationshipProcessorService.processFollow(job);
-				case 'unfollow': return this.relationshipProcessorService.processUnfollow(job);
-				case 'block': return this.relationshipProcessorService.processBlock(job);
-				case 'unblock': return this.relationshipProcessorService.processUnblock(job);
-				default: throw new Error(`unrecognized job type ${job.name} for relationship`);
-			}
-		}, {
-			...baseQueueOptions(this.config, QUEUE.RELATIONSHIP),
-			autorun: false,
-			concurrency: this.config.relationshipJobConcurrency ?? 16,
-			limiter: {
-				max: this.config.relationshipJobPerSec ?? 64,
-				duration: 1000,
-			},
-		});
+		{
+			const processer = (job: Bull.Job) => {
+				switch (job.name) {
+					case 'follow': return this.relationshipProcessorService.processFollow(job);
+					case 'unfollow': return this.relationshipProcessorService.processUnfollow(job);
+					case 'block': return this.relationshipProcessorService.processBlock(job);
+					case 'unblock': return this.relationshipProcessorService.processUnblock(job);
+					default: throw new Error(`unrecognized job type ${job.name} for relationship`);
+				}
+			};
+
+			this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: Relationship: ' + job.name }, () => processer(job));
+				} else {
+					return processer(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.RELATIONSHIP),
+				autorun: false,
+				concurrency: this.config.relationshipJobConcurrency ?? 16,
+				limiter: {
+					max: this.config.relationshipJobPerSec ?? 64,
+					duration: 1000,
+				},
+			});
 
-		const relationshipLogger = this.logger.createSubLogger('relationship');
+			const logger = this.logger.createSubLogger('relationship');
 
-		this.relationshipQueueWorker
-			.on('active', (job) => relationshipLogger.debug(`active id=${job.id}`))
-			.on('completed', (job, result) => relationshipLogger.debug(`completed(${result}) id=${job.id}`))
-			.on('failed', (job, err) => relationshipLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
-			.on('error', (err: Error) => relationshipLogger.error(`error ${err.stack}`, { e: renderError(err) }))
-			.on('stalled', (jobId) => relationshipLogger.warn(`stalled id=${jobId}`));
+			this.relationshipQueueWorker
+				.on('active', (job) => logger.debug(`active id=${job.id}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+				.on('failed', (job, err) => {
+					logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: Relationship: ${job?.name ?? '?'}: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
 		//#endregion
 
 		//#region object storage
-		this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => {
-			switch (job.name) {
-				case 'deleteFile': return this.deleteFileProcessorService.process(job);
-				case 'cleanRemoteFiles': return this.cleanRemoteFilesProcessorService.process(job);
-				default: throw new Error(`unrecognized job type ${job.name} for objectStorage`);
-			}
-		}, {
-			...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE),
-			autorun: false,
-			concurrency: 16,
-		});
+		{
+			const processer = (job: Bull.Job) => {
+				switch (job.name) {
+					case 'deleteFile': return this.deleteFileProcessorService.process(job);
+					case 'cleanRemoteFiles': return this.cleanRemoteFilesProcessorService.process(job);
+					default: throw new Error(`unrecognized job type ${job.name} for objectStorage`);
+				}
+			};
+
+			this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: ObjectStorage: ' + job.name }, () => processer(job));
+				} else {
+					return processer(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE),
+				autorun: false,
+				concurrency: 16,
+			});
 
-		const objectStorageLogger = this.logger.createSubLogger('objectStorage');
+			const logger = this.logger.createSubLogger('objectStorage');
 
-		this.objectStorageQueueWorker
-			.on('active', (job) => objectStorageLogger.debug(`active id=${job.id}`))
-			.on('completed', (job, result) => objectStorageLogger.debug(`completed(${result}) id=${job.id}`))
-			.on('failed', (job, err) => objectStorageLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
-			.on('error', (err: Error) => objectStorageLogger.error(`error ${err.stack}`, { e: renderError(err) }))
-			.on('stalled', (jobId) => objectStorageLogger.warn(`stalled id=${jobId}`));
+			this.objectStorageQueueWorker
+				.on('active', (job) => logger.debug(`active id=${job.id}`))
+				.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+				.on('failed', (job, err) => {
+					logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+					if (config.sentryForBackend) {
+						Sentry.captureMessage(`Queue: ObjectStorage: ${job?.name ?? '?'}: ${err.message}`, {
+							level: 'error',
+							extra: { job, err },
+						});
+					}
+				})
+				.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+				.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+		}
 		//#endregion
 
 		//#region ended poll notification
-		this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => this.endedPollNotificationProcessorService.process(job), {
-			...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
-			autorun: false,
-		});
+		{
+			this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => {
+				if (this.config.sentryForBackend) {
+					return Sentry.startSpan({ name: 'Queue: EndedPollNotification' }, () => this.endedPollNotificationProcessorService.process(job));
+				} else {
+					return this.endedPollNotificationProcessorService.process(job);
+				}
+			}, {
+				...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
+				autorun: false,
+			});
+		}
 		//#endregion
 	}
 
@@ -338,7 +510,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
 			this.dbQueueWorker.run(),
 			this.deliverQueueWorker.run(),
 			this.inboxQueueWorker.run(),
-			this.webhookDeliverQueueWorker.run(),
+			this.userWebhookDeliverQueueWorker.run(),
+			this.systemWebhookDeliverQueueWorker.run(),
 			this.relationshipQueueWorker.run(),
 			this.objectStorageQueueWorker.run(),
 			this.endedPollNotificationQueueWorker.run(),
@@ -352,7 +525,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
 			this.dbQueueWorker.close(),
 			this.deliverQueueWorker.close(),
 			this.inboxQueueWorker.close(),
-			this.webhookDeliverQueueWorker.close(),
+			this.userWebhookDeliverQueueWorker.close(),
+			this.systemWebhookDeliverQueueWorker.close(),
 			this.relationshipQueueWorker.close(),
 			this.objectStorageQueueWorker.close(),
 			this.endedPollNotificationQueueWorker.close(),
diff --git a/packages/backend/src/queue/const.ts b/packages/backend/src/queue/const.ts
index 132e916612..67f689b618 100644
--- a/packages/backend/src/queue/const.ts
+++ b/packages/backend/src/queue/const.ts
@@ -14,7 +14,8 @@ export const QUEUE = {
 	DB: 'db',
 	RELATIONSHIP: 'relationship',
 	OBJECT_STORAGE: 'objectStorage',
-	WEBHOOK_DELIVER: 'webhookDeliver',
+	USER_WEBHOOK_DELIVER: 'userWebhookDeliver',
+	SYSTEM_WEBHOOK_DELIVER: 'systemWebhookDeliver',
 };
 
 export function baseQueueOptions(config: Config, queueName: typeof QUEUE[keyof typeof QUEUE]): Bull.QueueOptions {
diff --git a/packages/backend/src/queue/processors/DeliverProcessorService.ts b/packages/backend/src/queue/processors/DeliverProcessorService.ts
index b73195afc3..d665945861 100644
--- a/packages/backend/src/queue/processors/DeliverProcessorService.ts
+++ b/packages/backend/src/queue/processors/DeliverProcessorService.ts
@@ -109,6 +109,12 @@ export class DeliverProcessorService {
 							suspensionState: 'autoSuspendedForNotResponding',
 						});
 					}
+				} else {
+					// isNotRespondingがtrueでnotRespondingSinceがnullの場合はnotRespondingSinceをセット
+					// notRespondingSinceは新たな機能なので、それ以前のデータにはnotRespondingSinceがない場合がある
+					this.federatedInstanceService.update(i.id, {
+						notRespondingSince: new Date(),
+					});
 				}
 
 				this.apRequestChart.deliverFail();
diff --git a/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts b/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts
index 29c1f27bb1..34180e5f2b 100644
--- a/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts
+++ b/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts
@@ -7,6 +7,7 @@ import { Inject, Injectable } from '@nestjs/common';
 import { DI } from '@/di-symbols.js';
 import type { PollVotesRepository, NotesRepository } from '@/models/_.js';
 import type Logger from '@/logger.js';
+import { CacheService } from '@/core/CacheService.js';
 import { NotificationService } from '@/core/NotificationService.js';
 import { bindThis } from '@/decorators.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
@@ -24,6 +25,7 @@ export class EndedPollNotificationProcessorService {
 		@Inject(DI.pollVotesRepository)
 		private pollVotesRepository: PollVotesRepository,
 
+		private cacheService: CacheService,
 		private notificationService: NotificationService,
 		private queueLoggerService: QueueLoggerService,
 	) {
@@ -47,9 +49,12 @@ export class EndedPollNotificationProcessorService {
 		const userIds = [...new Set([note.userId, ...votes.map(v => v.userId)])];
 
 		for (const userId of userIds) {
-			this.notificationService.createNotification(userId, 'pollEnded', {
-				noteId: note.id,
-			});
+			const profile = await this.cacheService.userProfileCache.fetch(userId);
+			if (profile.userHost === null) {
+				this.notificationService.createNotification(userId, 'pollEnded', {
+					noteId: note.id,
+				});
+			}
 		}
 	}
 }
diff --git a/packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts b/packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts
new file mode 100644
index 0000000000..f6bef52684
--- /dev/null
+++ b/packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts
@@ -0,0 +1,87 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Inject, Injectable } from '@nestjs/common';
+import * as Bull from 'bullmq';
+import { DI } from '@/di-symbols.js';
+import type { SystemWebhooksRepository } from '@/models/_.js';
+import type { Config } from '@/config.js';
+import type Logger from '@/logger.js';
+import { HttpRequestService } from '@/core/HttpRequestService.js';
+import { StatusError } from '@/misc/status-error.js';
+import { bindThis } from '@/decorators.js';
+import { QueueLoggerService } from '../QueueLoggerService.js';
+import { SystemWebhookDeliverJobData } from '../types.js';
+
+@Injectable()
+export class SystemWebhookDeliverProcessorService {
+	private logger: Logger;
+
+	constructor(
+		@Inject(DI.config)
+		private config: Config,
+
+		@Inject(DI.systemWebhooksRepository)
+		private systemWebhooksRepository: SystemWebhooksRepository,
+
+		private httpRequestService: HttpRequestService,
+		private queueLoggerService: QueueLoggerService,
+	) {
+		this.logger = this.queueLoggerService.logger.createSubLogger('webhook');
+	}
+
+	@bindThis
+	public async process(job: Bull.Job<SystemWebhookDeliverJobData>): Promise<string> {
+		try {
+			this.logger.debug(`delivering ${job.data.webhookId}`);
+
+			const res = await this.httpRequestService.send(job.data.to, {
+				method: 'POST',
+				headers: {
+					'User-Agent': 'Misskey-Hooks',
+					'X-Misskey-Host': this.config.host,
+					'X-Misskey-Hook-Id': job.data.webhookId,
+					'X-Misskey-Hook-Secret': job.data.secret,
+					'Content-Type': 'application/json',
+				},
+				body: JSON.stringify({
+					server: this.config.url,
+					hookId: job.data.webhookId,
+					eventId: job.data.eventId,
+					createdAt: job.data.createdAt,
+					type: job.data.type,
+					body: job.data.content,
+				}),
+			});
+
+			this.systemWebhooksRepository.update({ id: job.data.webhookId }, {
+				latestSentAt: new Date(),
+				latestStatus: res.status,
+			});
+
+			return 'Success';
+		} catch (res) {
+			this.logger.error(res as Error);
+
+			this.systemWebhooksRepository.update({ id: job.data.webhookId }, {
+				latestSentAt: new Date(),
+				latestStatus: res instanceof StatusError ? res.statusCode : 1,
+			});
+
+			if (res instanceof StatusError) {
+				// 4xx
+				if (!res.isRetryable) {
+					throw new Bull.UnrecoverableError(`${res.statusCode} ${res.statusMessage}`);
+				}
+
+				// 5xx etc.
+				throw new Error(`${res.statusCode} ${res.statusMessage}`);
+			} else {
+				// DNS error, socket error, timeout ...
+				throw res;
+			}
+		}
+	}
+}
diff --git a/packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts b/packages/backend/src/queue/processors/UserWebhookDeliverProcessorService.ts
index 8c260c0137..9ec630ef70 100644
--- a/packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts
+++ b/packages/backend/src/queue/processors/UserWebhookDeliverProcessorService.ts
@@ -13,10 +13,10 @@ import { HttpRequestService } from '@/core/HttpRequestService.js';
 import { StatusError } from '@/misc/status-error.js';
 import { bindThis } from '@/decorators.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
-import type { WebhookDeliverJobData } from '../types.js';
+import { UserWebhookDeliverJobData } from '../types.js';
 
 @Injectable()
-export class WebhookDeliverProcessorService {
+export class UserWebhookDeliverProcessorService {
 	private logger: Logger;
 
 	constructor(
@@ -33,7 +33,7 @@ export class WebhookDeliverProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<WebhookDeliverJobData>): Promise<string> {
+	public async process(job: Bull.Job<UserWebhookDeliverJobData>): Promise<string> {
 		try {
 			this.logger.debug(`delivering ${job.data.webhookId}`);
 
diff --git a/packages/backend/src/queue/types.ts b/packages/backend/src/queue/types.ts
index ce57ba745e..a4077a0547 100644
--- a/packages/backend/src/queue/types.ts
+++ b/packages/backend/src/queue/types.ts
@@ -106,7 +106,17 @@ export type EndedPollNotificationJobData = {
 	noteId: MiNote['id'];
 };
 
-export type WebhookDeliverJobData = {
+export type SystemWebhookDeliverJobData = {
+	type: string;
+	content: unknown;
+	webhookId: MiWebhook['id'];
+	to: string;
+	secret: string;
+	createdAt: number;
+	eventId: string;
+};
+
+export type UserWebhookDeliverJobData = {
 	type: string;
 	content: unknown;
 	webhookId: MiWebhook['id'];