Bump version to 2026.7.0-alpha.4

* feat: BullMQのジョブ追加時に発動する手動計測の追加

* add comment

.config/example.yml

@@ -357,6 +357,10 @@ id: 'aidx'
 #  #resourceAttributes:
 #  #  deployment.environment: 'production'
 #  #propagateTraceToRemote: false
+#  # Queue worker spans are independent traces linked to their enqueuer by default.
+#  # Set to 'parent' to make them child spans instead. This can create very large
+#  # traces for high-fan-out queues such as federation delivery.
+#  #jobTraceContextMode: 'link'
 
 #sentryForFrontend:
 #  vueIntegration:

CHANGELOG.md

@@ -39,6 +39,8 @@
 ### Server
 - Enhance: センシティブメディアの判定を外部サービス ([sensitive-detector](https://github.com/misskey-dev/sensitive-detector)) に分離し、`nsfwjs` / `@tensorflow/tfjs(-node)` の同梱と NSFW 判定モデルを廃止 (#16804)
 - Enhance: バックエンドの `otelForBackend` 設定で OpenTelemetry Traces を OTLP Collector に送信できるように
+- Enhance: OpenTelemetry を単独で利用する際、外向き HTTP リクエストを自動計装するように
+- Enhance: OpenTelemetry 利用時に BullMQ ジョブを enqueue 元の trace と Span Link で関連付けられるように
 - Enhance: Sentry バックエンドの自動計装を `sentryForBackend.disabledIntegrations` で個別に無効化できるように
 - Enhance: Node.js 22.23.0以降、24.17.0以降、26.4.0以降をサポートするように
 - Enhance: Docker Image の Node.js を 26.4.0 に、Debian を trixie (v13) に更新

package.json

@@ -1,6 +1,6 @@
 {
 	"name": "misskey",
-	"version": "2026.7.0-alpha.3",
+	"version": "2026.7.0-alpha.4",
 	"codename": "nasubi",
 	"repository": {
 		"type": "git",

packages/backend/src/config.ts

@@ -32,6 +32,7 @@ type OtelBackendConfig = {
 	sampleRate?: number;
 	resourceAttributes?: Record<string, string>;
 	propagateTraceToRemote?: boolean;
+	jobTraceContextMode?: 'link' | 'parent';
 };
 
 /**

packages/backend/src/core/QueueModule.ts

@@ -9,6 +9,7 @@ import { DI } from '@/di-symbols.js';
 import type { Config } from '@/config.js';
 import { baseQueueOptions, QUEUE } from '@/queue/const.js';
 import { allSettled } from '@/misc/promise-tracker.js';
+import { instrumentQueue } from '@/core/telemetry/queue-instrumentation.js';
 import {
 	DeliverJobData,
 	EndedPollNotificationJobData,
@@ -31,63 +32,72 @@ export type ObjectStorageQueue = Bull.Queue;
 export type UserWebhookDeliverQueue = Bull.Queue<UserWebhookDeliverJobData>;
 export type SystemWebhookDeliverQueue = Bull.Queue<SystemWebhookDeliverJobData>;
 
+function createQueue<T extends object>(queueName: string, config: Config): Bull.Queue<T> {
+	const queue = new Bull.Queue<T>(queueName, baseQueueOptions(config, queueName));
+	// Queue のラップは、enqueue 時に OTel context をジョブデータへ埋め込むためのもの。
+	// Sentry 単独ではジョブ間の context 伝播を使わないので、OTel 未設定時は元の Queue を返す。
+	if (config.otelForBackend == null) return queue;
+
+	return instrumentQueue(queue);
+}
+
 const $system: Provider = {
 	provide: 'queue:system',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.SYSTEM, baseQueueOptions(config, QUEUE.SYSTEM)),
+	useFactory: (config: Config) => createQueue<Record<string, unknown>>(QUEUE.SYSTEM, config),
 	inject: [DI.config],
 };
 
 const $endedPollNotification: Provider = {
 	provide: 'queue:endedPollNotification',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.ENDED_POLL_NOTIFICATION, baseQueueOptions(config, QUEUE.ENDED_POLL_NOTIFICATION)),
+	useFactory: (config: Config) => createQueue<EndedPollNotificationJobData>(QUEUE.ENDED_POLL_NOTIFICATION, config),
 	inject: [DI.config],
 };
 
 const $postScheduledNote: Provider = {
 	provide: 'queue:postScheduledNote',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.POST_SCHEDULED_NOTE, baseQueueOptions(config, QUEUE.POST_SCHEDULED_NOTE)),
+	useFactory: (config: Config) => createQueue<PostScheduledNoteJobData>(QUEUE.POST_SCHEDULED_NOTE, config),
 	inject: [DI.config],
 };
 
 const $deliver: Provider = {
 	provide: 'queue:deliver',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER)),
+	useFactory: (config: Config) => createQueue<DeliverJobData>(QUEUE.DELIVER, config),
 	inject: [DI.config],
 };
 
 const $inbox: Provider = {
 	provide: 'queue:inbox',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.INBOX, baseQueueOptions(config, QUEUE.INBOX)),
+	useFactory: (config: Config) => createQueue<InboxJobData>(QUEUE.INBOX, config),
 	inject: [DI.config],
 };
 
 const $db: Provider = {
 	provide: 'queue:db',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.DB, baseQueueOptions(config, QUEUE.DB)),
+	useFactory: (config: Config) => createQueue<Record<string, unknown>>(QUEUE.DB, config),
 	inject: [DI.config],
 };
 
 const $relationship: Provider = {
 	provide: 'queue:relationship',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.RELATIONSHIP, baseQueueOptions(config, QUEUE.RELATIONSHIP)),
+	useFactory: (config: Config) => createQueue<RelationshipJobData>(QUEUE.RELATIONSHIP, config),
 	inject: [DI.config],
 };
 
 const $objectStorage: Provider = {
 	provide: 'queue:objectStorage',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.OBJECT_STORAGE, baseQueueOptions(config, QUEUE.OBJECT_STORAGE)),
+	useFactory: (config: Config) => createQueue<Record<string, unknown>>(QUEUE.OBJECT_STORAGE, config),
 	inject: [DI.config],
 };
 
 const $userWebhookDeliver: Provider = {
 	provide: 'queue:userWebhookDeliver',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.USER_WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.USER_WEBHOOK_DELIVER)),
+	useFactory: (config: Config) => createQueue<UserWebhookDeliverJobData>(QUEUE.USER_WEBHOOK_DELIVER, config),
 	inject: [DI.config],
 };
 
 const $systemWebhookDeliver: Provider = {
 	provide: 'queue:systemWebhookDeliver',
-	useFactory: (config: Config) => new Bull.Queue(QUEUE.SYSTEM_WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.SYSTEM_WEBHOOK_DELIVER)),
+	useFactory: (config: Config) => createQueue<SystemWebhookDeliverJobData>(QUEUE.SYSTEM_WEBHOOK_DELIVER, config),
 	inject: [DI.config],
 };
 

packages/backend/src/core/telemetry/TelemetryService.ts

@@ -5,7 +5,7 @@
 
 import { Injectable } from '@nestjs/common';
 import { bindThis } from '@/decorators.js';
-import { captureMessage, shutdownTelemetry, startSpan } from './telemetry-registry.js';
+import { captureMessage, shutdownTelemetry, startSpan, startSpanWithTraceContext } from './telemetry-registry.js';
 import type { OnApplicationShutdown } from '@nestjs/common';
 import type { TelemetryCaptureMessageOptions } from './adapters/TelemetryAdapter.js';
 
@@ -21,6 +21,13 @@ export class TelemetryService implements OnApplicationShutdown {
 		return startSpan(name, fn);
 	}
 
+	@bindThis
+	public startSpanWithTraceContext<T>(name: string, jobData: object, fn: () => T): T {
+		// jobData に enqueue 元の context があれば worker span へ復元する。
+		// context の無い既存ジョブは、通常の startSpan と同じ扱いになる。
+		return startSpanWithTraceContext(name, jobData, fn);
+	}
+
 	@bindThis
 	public async onApplicationShutdown(_signal?: string): Promise<void> {
 		await shutdownTelemetry();

packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts

@@ -7,10 +7,13 @@ import * as os from 'node:os';
 import cluster from 'node:cluster';
 import { envOption } from '@/env.js';
 import { registerDiagLogger } from '@/core/telemetry/telemetry-diag.js';
+import { installHttpClientInstrumentation } from '@/core/telemetry/http-client-instrumentation.js';
+import { executeSpan, getQueueTraceContextMode, injectActiveTraceContext, recordSpanError, startSpanWithQueueTraceContext } from '@/core/telemetry/queue-trace-context.js';
 import type { Span, SpanStatusCode, Tracer } from '@opentelemetry/api';
 import type { Resource, ResourceDetector } from '@opentelemetry/resources';
 import type { ParentBasedSampler, Sampler } from '@opentelemetry/sdk-trace-base';
 import type { OtelBackendRuntimeConfig, TelemetryAdapter, TelemetryCaptureMessageOptions } from './TelemetryAdapter.js';
+import type { QueueTraceContextCarrier, QueueTraceContextDeps } from '../queue-trace-context.js';
 
 const DEFAULT_SHUTDOWN_TIMEOUT = 5000;
 
@@ -22,6 +25,8 @@ type OpenTelemetryAdapterDeps = {
 	getActiveSpan: () => Span | undefined;
 	spanStatusCodeError: SpanStatusCode;
 	shutdownTimeout: number;
+	shutdownHttpClientInstrumentation?: () => void;
+	queueTraceContext?: QueueTraceContextDeps;
 };
 
 type CreateSamplerDeps = {
@@ -48,7 +53,7 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 
 	public static async create(config: OtelBackendRuntimeConfig): Promise<OpenTelemetryAdapter> {
 		const [
-			{ diag, DiagLogLevel, SpanStatusCode, trace },
+			{ context, diag, DiagLogLevel, propagation, ROOT_CONTEXT, SpanKind, SpanStatusCode, trace },
 			{ W3CTraceContextPropagator },
 			{ OTLPTraceExporter },
 			{ defaultResource, detectResources, envDetector, resourceFromAttributes },
@@ -100,12 +105,28 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 		});
 
 		// provider操作をdepsに閉じ込め、span wrapper本体をユニットテストしやすくする。
+		const tracer = provider.getTracer('misskey-backend');
+
 		return new OpenTelemetryAdapter({
-			tracer: provider.getTracer('misskey-backend'),
+			tracer,
 			provider,
 			getActiveSpan: () => trace.getActiveSpan(),
 			spanStatusCodeError: SpanStatusCode.ERROR,
 			shutdownTimeout: DEFAULT_SHUTDOWN_TIMEOUT,
+			shutdownHttpClientInstrumentation: installHttpClientInstrumentation({
+				tracer,
+				spanKindClient: SpanKind.CLIENT,
+				spanStatusCodeError: SpanStatusCode.ERROR,
+			}),
+			queueTraceContext: {
+				tracer,
+				propagation,
+				trace,
+				getActiveContext: () => context.active(),
+				rootContext: ROOT_CONTEXT,
+				mode: getQueueTraceContextMode(config.jobTraceContextMode),
+				spanStatusCodeError: SpanStatusCode.ERROR,
+			},
 		});
 	}
 
@@ -115,48 +136,38 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 		// 通知を握り潰さないよう、報告専用の短命spanを作ってそこに記録する。
 		const span = this.deps.getActiveSpan();
 		if (span != null) {
-			recordError(span, new Error(message), this.deps.spanStatusCodeError);
+			recordSpanError(span, new Error(message), this.deps.spanStatusCodeError);
 			return;
 		}
 
 		this.deps.tracer.startActiveSpan('captureMessage', reportSpan => {
-			recordError(reportSpan, new Error(message), this.deps.spanStatusCodeError);
+			recordSpanError(reportSpan, new Error(message), this.deps.spanStatusCodeError);
 			reportSpan.end();
 		});
 	}
 
 	public startSpan<T>(name: string, fn: () => T): T {
 		// 既存のTelemetryAdapter契約に合わせ、同期/非同期どちらでも同じspan lifetimeを保証する。
-		return this.deps.tracer.startActiveSpan(name, span => {
-			try {
-				const result = fn();
-				if (isPromiseLike(result)) {
-					// Promiseの場合は完了/失敗までspanを開いたままにし、失敗時はerror statusへ変換する。
-					return result.then(
-						value => {
-							span.end();
-							return value;
-						},
-						error => {
-							recordError(span, error, this.deps.spanStatusCodeError);
-							span.end();
-							throw error;
-						},
-					) as T;
-				}
+		return this.deps.tracer.startActiveSpan(name, span => executeSpan(span, fn, this.deps.spanStatusCodeError));
+	}
 
-				// 同期成功はここでspanを閉じる。例外はcatch側で記録してから再throwする。
-				span.end();
-				return result;
-			} catch (error) {
-				recordError(span, error, this.deps.spanStatusCodeError);
-				span.end();
-				throw error;
-			}
-		});
+	public injectTraceContext(carrier: QueueTraceContextCarrier): void {
+		const queueTraceContext = this.deps.queueTraceContext;
+		// Queue context 用の依存は任意なので、無い場合はジョブデータを変更しない。
+		if (queueTraceContext == null) return;
+		injectActiveTraceContext(queueTraceContext, carrier);
+	}
+
+	public startSpanWithTraceContext<T>(name: string, jobData: object, fn: () => T): T {
+		const queueTraceContext = this.deps.queueTraceContext;
+		// Queue context 用の依存が無い場合は、従来の span 作成経路と同じ動作を保つ。
+		if (queueTraceContext == null) return this.startSpan(name, fn);
+
+		return startSpanWithQueueTraceContext(queueTraceContext, name, jobData, fn, () => this.startSpan(name, fn));
 	}
 
 	public async shutdown(): Promise<void> {
+		this.deps.shutdownHttpClientInstrumentation?.();
 		// BatchSpanProcessorのflushが詰まってもプロセス終了を妨げないよう、上限時間を設ける。
 		// タイムアウト側のtimerは、flushが先に終わった場合にイベントループを無駄に引き留めないようclearする。
 		let timer: NodeJS.Timeout | undefined;
@@ -201,20 +212,6 @@ export function createSampler(sampleRate: number, deps: CreateSamplerDeps): Pare
 	});
 }
 
-function recordError(span: Span, error: unknown, spanStatusCodeError: SpanStatusCode): void {
-	// throw値がError以外でもOTel exporterへ渡せる例外表現に正規化する。
-	const exception = error instanceof Error ? error : new Error(String(error));
-	span.recordException(exception);
-	span.setStatus({
-		code: spanStatusCodeError,
-		message: exception.message,
-	});
-}
-
-function isPromiseLike<T>(value: T): value is T & PromiseLike<Awaited<T>> {
-	return value != null && typeof (value as { then?: unknown }).then === 'function';
-}
-
 export function getMisskeyProcessRole(): string {
 	// Trace backend上でserver/queue/workerを見分けられるよう、Misskey固有の役割をresourceに載せる。
 	if (envOption.disableClustering) {

packages/backend/src/core/telemetry/adapters/SentryTelemetryAdapter.ts

@@ -5,9 +5,11 @@
 
 import Logger from '@/logger.js';
 import { registerDiagLogger } from '@/core/telemetry/telemetry-diag.js';
+import { getQueueTraceContextMode, injectActiveTraceContext, startSpanWithQueueTraceContext } from '@/core/telemetry/queue-trace-context.js';
 import type * as SentryNode from '@sentry/node';
 import type { NodeOptions } from '@sentry/node';
 import type { OtelBackendRuntimeConfig, SentryBackendConfig, TelemetryAdapter, TelemetryCaptureMessageOptions } from './TelemetryAdapter.js';
+import type { QueueTraceContextCarrier, QueueTraceContextDeps } from '../queue-trace-context.js';
 
 // OpenTelemetryAdapterのDEFAULT_SHUTDOWN_TIMEOUTと揃え、Sentryのtransportが詰まってもプロセス終了を妨げないようにする。
 const DEFAULT_SHUTDOWN_TIMEOUT = 5000;
@@ -113,6 +115,7 @@ export function buildSentryOtlpInitOptions(options: BuildSentryOtlpInitOptions):
 export class SentryTelemetryAdapter implements TelemetryAdapter {
 	private constructor(
 		private readonly Sentry: typeof SentryNode,
+		private readonly queueTraceContext?: QueueTraceContextDeps,
 	) {
 	}
 
@@ -131,7 +134,7 @@ export class SentryTelemetryAdapter implements TelemetryAdapter {
 	): Promise<SentryTelemetryAdapter> {
 		const Sentry = await import('@sentry/node');
 		const { nodeProfilingIntegration } = await import('@sentry/profiling-node');
-		const { diag, DiagLogLevel } = await import('@opentelemetry/api');
+		const { context, diag, DiagLogLevel, propagation, ROOT_CONTEXT, SpanStatusCode, trace } = await import('@opentelemetry/api');
 		const { BatchSpanProcessor } = await import('@opentelemetry/sdk-trace-base');
 		const { OTLPTraceExporter } = await import('@opentelemetry/exporter-trace-otlp-proto');
 
@@ -151,7 +154,17 @@ export class SentryTelemetryAdapter implements TelemetryAdapter {
 			nodeProfilingIntegration,
 		}));
 
-		return new SentryTelemetryAdapter(Sentry);
+		// Sentry が初期化した同じ OTel provider から tracer/context API を受け取り、
+		// Queue を跨ぐ context 伝播も Sentry と OTLP の両方へ同一 span として出力する。
+		return new SentryTelemetryAdapter(Sentry, {
+			tracer: trace.getTracer('misskey-backend'),
+			propagation,
+			trace,
+			getActiveContext: () => context.active(),
+			rootContext: ROOT_CONTEXT,
+			mode: getQueueTraceContextMode(otelConfig.jobTraceContextMode),
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		});
 	}
 
 	public captureMessage(message: string, opts: TelemetryCaptureMessageOptions): void {
@@ -166,6 +179,19 @@ export class SentryTelemetryAdapter implements TelemetryAdapter {
 		return this.Sentry.startSpan({ name }, fn);
 	}
 
+	public injectTraceContext(carrier: QueueTraceContextCarrier): void {
+		// Sentry 単体構成では queueTraceContext を持たず、従来どおりジョブデータを変更しない。
+		if (this.queueTraceContext == null) return;
+		injectActiveTraceContext(this.queueTraceContext, carrier);
+	}
+
+	public startSpanWithTraceContext<T>(name: string, jobData: object, fn: () => T): T {
+		// Sentry 単体構成では Sentry 既存の span 作成経路を使う。
+		if (this.queueTraceContext == null) return this.startSpan(name, fn);
+
+		return startSpanWithQueueTraceContext(this.queueTraceContext, name, jobData, fn, () => this.startSpan(name, fn));
+	}
+
 	public async shutdown(): Promise<void> {
 		// timeout未指定だとtransportのflushが詰まった際にプロセス終了を妨げるため、上限時間を設ける。
 		await this.Sentry.close(DEFAULT_SHUTDOWN_TIMEOUT);

packages/backend/src/core/telemetry/adapters/TelemetryAdapter.ts

@@ -4,6 +4,7 @@
  */
 
 import type { Config } from '@/config.js';
+import type { QueueTraceContextCarrier } from '../queue-trace-context.js';
 
 export type SentryBackendConfig = NonNullable<Config['sentryForBackend']>;
 export type OtelBackendConfig = NonNullable<Config['otelForBackend']>;
@@ -40,6 +41,18 @@ export interface TelemetryAdapter {
 	 */
 	startSpan<T>(name: string, fn: () => T): T;
 
+	/**
+	 * BullMQ のジョブデータへ保存する carrier に、active trace context を注入する。
+	 * OTel を使わない adapter は実装しない。
+	 */
+	injectTraceContext?(carrier: QueueTraceContextCarrier): void;
+
+	/**
+	 * ジョブに保存された enqueue 元の context を、worker span の Link または parent として復元する。
+	 * context を持たないジョブの互換性は adapter 側で保つ。
+	 */
+	startSpanWithTraceContext?<T>(name: string, jobData: object, fn: () => T): T;
+
 	/**
 	 * プロセス終了時にtelemetry backendへ残りのデータをflushする。
 	 * 実装側ではtransport停止に引きずられないよう、待機時間に上限を設ける。

packages/backend/src/core/telemetry/http-client-instrumentation.ts

@@ -0,0 +1,122 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { channel } from 'node:diagnostics_channel';
+import type { ClientRequest, IncomingMessage } from 'node:http';
+import type { Span, SpanOptions, SpanStatusCode, Tracer } from '@opentelemetry/api';
+
+const HTTP_CLIENT_REQUEST_CREATED = 'http.client.request.created';
+const HTTP_CLIENT_RESPONSE_FINISH = 'http.client.response.finish';
+const HTTP_CLIENT_REQUEST_ERROR = 'http.client.request.error';
+
+type HttpClientSpan = Pick<Span, 'end' | 'recordException' | 'setAttribute' | 'setStatus'>;
+
+type HttpClientInstrumentationDeps = {
+	tracer: Pick<Tracer, 'startSpan'>;
+	spanKindClient: SpanOptions['kind'];
+	spanStatusCodeError: SpanStatusCode;
+	subscribe: (name: string, listener: (message: unknown) => void) => () => void;
+};
+
+type RequestCreatedMessage = { request: ClientRequest };
+type ResponseFinishMessage = { request: ClientRequest; response: IncomingMessage };
+type RequestErrorMessage = { request: ClientRequest; error: Error };
+
+/**
+ * require フックを使わず、Node.js 組み込み HTTP クライアントの diagnostics channel を計装する。
+ * telemetry 初期化前に読み込まれたモジュールも対象になる。
+ */
+export function createHttpClientInstrumentation(deps: HttpClientInstrumentationDeps): () => void {
+	const spans = new WeakMap<ClientRequest, HttpClientSpan>();
+
+	const unsubscribeCreated = deps.subscribe(HTTP_CLIENT_REQUEST_CREATED, (message: unknown) => {
+		const { request } = message as RequestCreatedMessage;
+		const { url, host, port } = getRequestDetails(request);
+		const method = request.method ?? 'GET';
+		const span = deps.tracer.startSpan(method, {
+			kind: deps.spanKindClient,
+			attributes: {
+				'http.request.method': method,
+				'url.full': url,
+				'server.address': host,
+				'server.port': port,
+			},
+		});
+		spans.set(request, span);
+	});
+
+	const unsubscribeResponseFinish = deps.subscribe(HTTP_CLIENT_RESPONSE_FINISH, (message: unknown) => {
+		const { request, response } = message as ResponseFinishMessage;
+		const span = spans.get(request);
+		if (span == null) return;
+
+		const statusCode = response.statusCode;
+		if (statusCode != null) {
+			span.setAttribute('http.response.status_code', statusCode);
+		}
+		if (response.httpVersion != null) {
+			span.setAttribute('network.protocol.version', response.httpVersion);
+		}
+		if (statusCode != null && statusCode >= 400) {
+			span.setAttribute('error.type', String(statusCode));
+			span.setStatus({ code: deps.spanStatusCodeError });
+		}
+		span.end();
+		spans.delete(request);
+	});
+
+	const unsubscribeRequestError = deps.subscribe(HTTP_CLIENT_REQUEST_ERROR, (message: unknown) => {
+		const { request, error } = message as RequestErrorMessage;
+		const span = spans.get(request);
+		if (span == null) return;
+
+		span.recordException(error);
+		span.setAttribute('error.type', getErrorType(error));
+		span.setStatus({ code: deps.spanStatusCodeError });
+		span.end();
+		spans.delete(request);
+	});
+
+	return () => {
+		unsubscribeCreated();
+		unsubscribeResponseFinish();
+		unsubscribeRequestError();
+	};
+}
+
+export function installHttpClientInstrumentation(deps: Omit<HttpClientInstrumentationDeps, 'subscribe'>): () => void {
+	return createHttpClientInstrumentation({
+		...deps,
+		subscribe: (name, listener) => {
+			const diagnosticChannel = channel(name);
+			diagnosticChannel.subscribe(listener);
+			return () => diagnosticChannel.unsubscribe(listener);
+		},
+	});
+}
+
+function getRequestDetails(request: ClientRequest): { url: string; host: string; port: number } {
+	const protocol = request.protocol ?? 'http:';
+	const host = request.getHeader('host')?.toString() ?? request.host ?? 'localhost';
+	const url = new URL(request.path || '/', `${protocol}//${host}`);
+	// URL 属性には認証情報やクエリ文字列を含めない。
+	url.username = '';
+	url.password = '';
+	url.search = '';
+	url.hash = '';
+
+	return {
+		url: url.toString(),
+		host: url.hostname,
+		// URL.port は既定ポートでは空文字列になるため、スキームから補う。
+		port: url.port === '' ? (url.protocol === 'https:' ? 443 : 80) : Number(url.port),
+	};
+}
+
+function getErrorType(error: Error): string {
+	// Node.js の system error code は安定した低カーディナリティの識別子になる。
+	const code = (error as NodeJS.ErrnoException).code;
+	return code ?? error.name;
+}

packages/backend/src/core/telemetry/queue-instrumentation.ts

@@ -0,0 +1,33 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { injectTraceContext } from './telemetry-registry.js';
+import { injectQueueTraceContext } from './queue-trace-context.js';
+import type * as Bull from 'bullmq';
+
+/**
+ * Queue の add/addBulk をラップし、全ての BullMQ enqueue 経路を一箇所で捕捉する。
+ * QueueService を通さず直接 add/addBulk する呼び出し元もあるため、それぞれで注入すると漏れやすい。
+ */
+export function instrumentQueue<T extends object>(queue: Bull.Queue<T>): Bull.Queue<T> {
+	// BullMQ のメソッドは Queue インスタンスを this として使うため、差し替え前に bind して保持する。
+	const add = queue.add.bind(queue);
+	queue.add = ((name, data, opts) => {
+		// BullMQ が data を Redis 用にシリアライズする前に、enqueue 元の context を内部フィールドへ追加する。
+		injectQueueTraceContext(data, injectTraceContext);
+		return add(name, data, opts);
+	}) as typeof queue.add;
+
+	// addBulk は複数ジョブを一度にシリアライズするので、各 data へ同じ context を注入する。
+	const addBulk = queue.addBulk.bind(queue);
+	queue.addBulk = ((jobs) => {
+		for (const job of jobs) {
+			injectQueueTraceContext(job.data, injectTraceContext);
+		}
+		return addBulk(jobs);
+	}) as typeof queue.addBulk;
+
+	return queue;
+}

packages/backend/src/core/telemetry/queue-trace-context.ts

@@ -0,0 +1,182 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import type { Context, PropagationAPI, Span, SpanContext, SpanOptions, SpanStatusCode, Tracer } from '@opentelemetry/api';
+
+/*
+ * enqueue (push) から worker による取得・処理 (pop) までをテレメトリ上で関連付け、
+ * Queue を挟む非同期処理を一連の流れとして追跡できるようにする。
+ * そのため、trace context を次の流れで伝播する:
+ *
+ * 1. producer 側で active context を W3C Trace Context 形式の carrier へ注入する。
+ * 2. carrier をジョブデータの内部フィールドに保存し、BullMQ/Redis 経由で worker へ渡す。
+ * 3. worker 側で carrier を context へ戻し、設定に応じて Link または parent に使う。
+ *
+ * OTel API への依存を引数にしているのは、OTel 単体構成と Sentry + OTLP 構成で
+ * 同じ伝播処理を使うため。
+ */
+/** Redis 上のジョブデータにだけ保存する、OpenTelemetry propagator 用の carrier。 */
+export type QueueTraceContextCarrier = Record<string, string>;
+export type QueueTraceContextMode = 'link' | 'parent';
+
+// 通常のジョブプロセッサが参照しない Misskey 内部フィールドとして、ユーザー定義の data と区別する。
+const QUEUE_TRACE_CONTEXT_KEY = '__misskeyTraceContext';
+
+export type QueueTraceContextDeps = {
+	tracer: Pick<Tracer, 'startActiveSpan'>;
+	propagation: Pick<PropagationAPI, 'extract' | 'inject'>;
+	trace: {
+		getSpanContext(context: Context): SpanContext | undefined;
+	};
+	getActiveContext: () => Context;
+	rootContext: Context;
+	mode: QueueTraceContextMode;
+	spanStatusCodeError: SpanStatusCode;
+};
+
+export type QueueSpanContext = {
+	options: SpanOptions;
+	parentContext: Context;
+};
+
+type QueueSpanContextDeps = Pick<QueueTraceContextDeps, 'propagation' | 'trace' | 'rootContext' | 'mode'>;
+
+/**
+ * enqueue 元の active context を、ジョブ本来のデータを壊さない内部フィールドとして保持する。
+ * propagator が何も注入しなかった場合は、Redis に不要な空オブジェクトを残さない。
+ */
+export function injectQueueTraceContext(data: unknown, inject: (carrier: QueueTraceContextCarrier) => void): void {
+	// BullMQ の型定義外から渡される値も考慮し、書き込めない値は無視する。
+	if (data == null || typeof data !== 'object') return;
+
+	const carrier: QueueTraceContextCarrier = {};
+	inject(carrier);
+
+	// active span が無い場合、propagator は何も注入しない。空の内部フィールドは Redis に保存しない。
+	if (Object.keys(carrier).length === 0) return;
+	Object.assign(data, { [QUEUE_TRACE_CONTEXT_KEY]: carrier });
+}
+
+/** 現在実行中の span を propagator の標準形式で carrier へ書き出す。 */
+export function injectActiveTraceContext(deps: QueueTraceContextDeps, carrier: QueueTraceContextCarrier): void {
+	deps.propagation.inject(deps.getActiveContext(), carrier);
+}
+
+/**
+ * ジョブに保存された carrier から、worker span の開始に必要な context と options を組み立てる。
+ *
+ * - parent: enqueue span の子として同じ trace を継続する。
+ * - link: worker span を別の root trace にし、enqueue span への関連だけを Link に残す。
+ *
+ * link モードで trace を分けることで、worker 側の sampling 判定を enqueue 側から独立させられる。
+ */
+export function getQueueSpanContext(data: unknown, deps: QueueSpanContextDeps): QueueSpanContext | undefined {
+	const carrier = getQueueTraceContextCarrier(data);
+	if (carrier == null) return undefined;
+
+	// Redis から復元した carrier は別プロセス由来なので、現在の context ではなく ROOT_CONTEXT から展開する。
+	const extractedContext = deps.propagation.extract(deps.rootContext, carrier);
+	if (deps.mode === 'parent') {
+		return {
+			options: {},
+			parentContext: extractedContext,
+		};
+	}
+
+	// Link が受け取るのは Context ではなく SpanContext なので、extract 後に取り出す。
+	const spanContext = deps.trace.getSpanContext(extractedContext);
+	return {
+		options: {
+			root: true,
+			...(spanContext != null ? { links: [{ context: spanContext }] } : {}),
+		},
+		parentContext: deps.rootContext,
+	};
+}
+
+/**
+ * context を持つジョブは Link/parent の規則で span を開始する。
+ * デプロイ前に enqueue されたジョブなど、context を持たない場合は既存の adapter 固有実装へ委ねる。
+ */
+export function startSpanWithQueueTraceContext<T>(
+	deps: QueueTraceContextDeps,
+	name: string,
+	jobData: object,
+	fn: () => T,
+	fallback: () => T,
+): T {
+	const spanContext = getQueueSpanContext(jobData, deps);
+	if (spanContext == null) return fallback();
+
+	return deps.tracer.startActiveSpan(name, spanContext.options, spanContext.parentContext, span => executeSpan(span, fn, deps.spanStatusCodeError));
+}
+
+/**
+ * 既存の TelemetryAdapter 契約に合わせ、同期値と Promise のどちらを返す処理も span で包む。
+ * 成功・失敗のどちらでも処理完了まで span を開き、必ず一度だけ閉じる。
+ */
+export function executeSpan<T>(span: Span, fn: () => T, spanStatusCodeError: SpanStatusCode): T {
+	try {
+		const result = fn();
+		if (isPromiseLike(result)) {
+			// fn() の戻り値は T のまま保ちつつ、Promise の settle 時に span を閉じる。
+			return result.then(
+				value => {
+					span.end();
+					return value;
+				},
+				error => {
+					recordSpanError(span, error, spanStatusCodeError);
+					span.end();
+					throw error;
+				},
+			) as T;
+		}
+
+		span.end();
+		return result;
+	} catch (error) {
+		// fn() が同期的に throw した場合も、Promise の reject と同じ形で記録して呼び出し元へ戻す。
+		recordSpanError(span, error, spanStatusCodeError);
+		span.end();
+		throw error;
+	}
+}
+
+/** Error 以外の throw 値も OTel exporter が扱える例外に正規化し、span をエラー状態にする。 */
+export function recordSpanError(span: Span, error: unknown, spanStatusCodeError: SpanStatusCode): void {
+	const exception = error instanceof Error ? error : new Error(String(error));
+	span.recordException(exception);
+	span.setStatus({
+		code: spanStatusCodeError,
+		message: exception.message,
+	});
+}
+
+/**
+ * 未設定時は、enqueue と worker の sampling を分離できる link を使う。
+ * 設定ミスを無言でフォールバックさせないため、未知の値は起動時にエラーにする。
+ */
+export function getQueueTraceContextMode(mode: unknown): QueueTraceContextMode {
+	if (mode == null || mode === 'link') return 'link';
+	if (mode === 'parent') return 'parent';
+	throw new Error('otelForBackend.jobTraceContextMode must be either \'link\' or \'parent\'.');
+}
+
+function getQueueTraceContextCarrier(data: unknown): QueueTraceContextCarrier | undefined {
+	if (data == null || typeof data !== 'object') return undefined;
+	const carrier = (data as Record<string, unknown>)[QUEUE_TRACE_CONTEXT_KEY];
+	if (carrier == null || typeof carrier !== 'object' || Array.isArray(carrier)) return undefined;
+
+	// Redis 上の旧いジョブや壊れたデータで worker を落とさないよう、propagator に渡せる string map だけを受け入れる。
+	const entries = Object.entries(carrier);
+	if (entries.length === 0 || entries.some(([, value]) => typeof value !== 'string')) return undefined;
+	return Object.fromEntries(entries) as QueueTraceContextCarrier;
+}
+
+function isPromiseLike<T>(value: T): value is T & PromiseLike<Awaited<T>> {
+	// native Promise に限らず thenable も span の完了を待てるよう、instanceof ではなく then の有無で判定する。
+	return value != null && typeof (value as { then?: unknown }).then === 'function';
+}

packages/backend/src/core/telemetry/telemetry-registry.ts

@@ -7,6 +7,7 @@ import type { Config } from '@/config.js';
 import { OpenTelemetryAdapter } from './adapters/OpenTelemetryAdapter.js';
 import { SentryTelemetryAdapter } from './adapters/SentryTelemetryAdapter.js';
 import type { OtelBackendRuntimeConfig, TelemetryAdapter, TelemetryCaptureMessageOptions } from './adapters/TelemetryAdapter.js';
+import type { QueueTraceContextCarrier } from './queue-trace-context.js';
 
 /**
  * NestのDIコンテナが構築される前(boot処理内)で初期化する必要があるため、
@@ -54,6 +55,19 @@ export function startSpan<T>(name: string, fn: () => T): T {
 	return wrapped();
 }
 
+export function injectTraceContext(carrier: QueueTraceContextCarrier): void {
+	// Queue の carrier は共有データなので、通知と異なり全 adapter にブロードキャストしない。
+	// OTel provider は現在 1 つだけなので、同じ header を上書きしないよう最初の対応 adapter だけを使う。
+	adapters.find(adapter => adapter.injectTraceContext != null)?.injectTraceContext?.(carrier);
+}
+
+export function startSpanWithTraceContext<T>(name: string, jobData: object, fn: () => T): T {
+	// Queue context を解釈できる adapter に span 作成を任せる。
+	// 対応 adapter が無い構成では通常の startSpan へフォールバックする。
+	const adapter = adapters.find(adapter => adapter.startSpanWithTraceContext != null);
+	return adapter?.startSpanWithTraceContext?.(name, jobData, fn) ?? startSpan(name, fn);
+}
+
 export async function shutdownTelemetry(): Promise<void> {
 	// 終了時は登録済みadapterを並列にflush/shutdownする。
 	await Promise.allSettled(adapters.map(adapter => adapter.shutdown()));

packages/backend/src/queue/QueueProcessorService.ts

@@ -158,6 +158,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
 			};
 		}
 
+		// 以下の各 Worker は job.data に保存された enqueue 元の trace context を復元し、
+		// ジョブの実処理全体を Link または parent の worker span で囲む。
 		//#region system
 		{
 			const processer = (job: Bull.Job) => {
@@ -176,7 +178,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 			};
 
 			this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => {
-				return this.telemetryService.startSpan('Queue: System: ' + job.name, () => processer(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: System: ' + job.name, job.data, () => processer(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.SYSTEM),
 				autorun: false,
@@ -227,7 +229,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 			};
 
 			this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
-				return this.telemetryService.startSpan('Queue: DB: ' + job.name, () => processer(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: DB: ' + job.name, job.data, () => processer(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.DB),
 				autorun: false,
@@ -253,7 +255,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		//#region deliver
 		{
 			this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => {
-				return this.telemetryService.startSpan('Queue: Deliver', () => this.deliverProcessorService.process(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: Deliver', job.data, () => this.deliverProcessorService.process(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.DELIVER),
 				autorun: false,
@@ -287,7 +289,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		//#region inbox
 		{
 			this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => {
-				return this.telemetryService.startSpan('Queue: Inbox', () => this.inboxProcessorService.process(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: Inbox', job.data, () => this.inboxProcessorService.process(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.INBOX),
 				autorun: false,
@@ -321,7 +323,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		//#region user-webhook deliver
 		{
 			this.userWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.USER_WEBHOOK_DELIVER, (job) => {
-				return this.telemetryService.startSpan('Queue: UserWebhookDeliver', () => this.userWebhookDeliverProcessorService.process(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: UserWebhookDeliver', job.data, () => this.userWebhookDeliverProcessorService.process(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER),
 				autorun: false,
@@ -355,7 +357,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		//#region system-webhook deliver
 		{
 			this.systemWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.SYSTEM_WEBHOOK_DELIVER, (job) => {
-				return this.telemetryService.startSpan('Queue: SystemWebhookDeliver', () => this.systemWebhookDeliverProcessorService.process(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: SystemWebhookDeliver', job.data, () => this.systemWebhookDeliverProcessorService.process(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.SYSTEM_WEBHOOK_DELIVER),
 				autorun: false,
@@ -399,7 +401,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 			};
 
 			this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
-				return this.telemetryService.startSpan('Queue: Relationship: ' + job.name, () => processer(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: Relationship: ' + job.name, job.data, () => processer(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.RELATIONSHIP),
 				autorun: false,
@@ -438,7 +440,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 			};
 
 			this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => {
-				return this.telemetryService.startSpan('Queue: ObjectStorage: ' + job.name, () => processer(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: ObjectStorage: ' + job.name, job.data, () => processer(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.OBJECT_STORAGE),
 				autorun: false,
@@ -465,7 +467,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		//#region ended poll notification
 		{
 			this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => {
-				return this.telemetryService.startSpan('Queue: EndedPollNotification', () => this.endedPollNotificationProcessorService.process(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: EndedPollNotification', job.data, () => this.endedPollNotificationProcessorService.process(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
 				autorun: false,
@@ -476,7 +478,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		//#region post scheduled note
 		{
 			this.postScheduledNoteQueueWorker = new Bull.Worker(QUEUE.POST_SCHEDULED_NOTE, async (job) => {
-				return this.telemetryService.startSpan('Queue: PostScheduledNote', () => this.postScheduledNoteProcessorService.process(job));
+				return this.telemetryService.startSpanWithTraceContext('Queue: PostScheduledNote', job.data, () => this.postScheduledNoteProcessorService.process(job));
 			}, {
 				...baseWorkerOptions(this.config, QUEUE.POST_SCHEDULED_NOTE),
 				autorun: false,

packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts

@@ -8,6 +8,7 @@ import { SpanStatusCode } from '@opentelemetry/api';
 import { defaultResource, detectResources, envDetector, resourceFromAttributes } from '@opentelemetry/resources';
 import { ParentBasedSampler, TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-base';
 import { ATTR_SERVICE_INSTANCE_ID, ATTR_SERVICE_NAME, ATTR_SERVICE_VERSION } from '@opentelemetry/semantic-conventions';
+import type { Context, SpanContext } from '@opentelemetry/api';
 import { OpenTelemetryAdapter, createResource, createSampler, getMisskeyProcessRole } from '@/core/telemetry/adapters/OpenTelemetryAdapter.js';
 
 const mocks = vi.hoisted(() => {
@@ -97,6 +98,61 @@ describe('OpenTelemetryAdapter', () => {
 		expect(span.end).toHaveBeenCalledTimes(1);
 	});
 
+	test('creates a root worker span linked to the enqueue span by default', () => {
+		const span = {
+			end: vi.fn(),
+			recordException: vi.fn(),
+			setStatus: vi.fn(),
+		};
+		const rootContext = {} as Context;
+		const extractedContext = {} as Context;
+		const sourceSpanContext = {
+			traceId: '0123456789abcdef0123456789abcdef',
+			spanId: '0123456789abcdef',
+			traceFlags: 1,
+			isRemote: true,
+		} as SpanContext;
+		const propagation = {
+			isPropagationApi: true,
+			inject: vi.fn(),
+			extract(this: { isPropagationApi: boolean }) {
+				if (!this.isPropagationApi) throw new Error('lost propagation API receiver');
+				return extractedContext;
+			},
+		};
+		const tracer = {
+			startActiveSpan: vi.fn((_name: string, _options: unknown, _context: unknown, fn: (spanArg: typeof span) => string) => fn(span)),
+		} as any;
+		const adapter = new OpenTelemetryAdapter({
+			tracer,
+			provider: { shutdown: vi.fn() },
+			getActiveSpan: () => undefined,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+			shutdownTimeout: 10,
+			queueTraceContext: {
+				tracer,
+				propagation: propagation as any,
+				trace: { getSpanContext: () => sourceSpanContext },
+				getActiveContext: () => rootContext,
+				rootContext,
+				mode: 'link',
+				spanStatusCodeError: SpanStatusCode.ERROR,
+			},
+		});
+
+		expect(adapter.startSpanWithTraceContext('Queue: Deliver', {
+			__misskeyTraceContext: {
+				traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01',
+			},
+		}, () => 'ok')).toBe('ok');
+
+		expect(tracer.startActiveSpan).toHaveBeenCalledWith('Queue: Deliver', {
+			root: true,
+			links: [{ context: sourceSpanContext }],
+		}, rootContext, expect.any(Function));
+		expect(span.end).toHaveBeenCalledTimes(1);
+	});
+
 	test('bridges captureMessage to the active span when one exists', () => {
 		const activeSpan = {
 			recordException: vi.fn(),

packages/backend/test/unit/core/telemetry/adapters/SentryTelemetryAdapter.ts

@@ -208,8 +208,13 @@ describe('SentryTelemetryAdapter.createWithOtlpExport', () => {
 			nodeProfilingIntegration,
 		}));
 		vi.doMock('@opentelemetry/api', () => ({
+			context: { active: vi.fn() },
 			diag: { setLogger },
 			DiagLogLevel: { WARN: 50 },
+			propagation: { inject: vi.fn(), extract: vi.fn() },
+			ROOT_CONTEXT: {},
+			SpanStatusCode: { ERROR: 2 },
+			trace: { getTracer: vi.fn(), getSpanContext: vi.fn() },
 		}));
 		vi.doMock('@opentelemetry/sdk-trace-base', () => ({
 			BatchSpanProcessor,

packages/backend/test/unit/core/telemetry/http-client-instrumentation.ts

@@ -0,0 +1,108 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { describe, expect, test, vi } from 'vitest';
+import { SpanKind, SpanStatusCode } from '@opentelemetry/api';
+import { createHttpClientInstrumentation } from '@/core/telemetry/http-client-instrumentation.js';
+
+function request() {
+	return {
+		method: 'POST',
+		protocol: 'https:',
+		path: '/inbox?token=secret',
+		host: 'remote.example',
+		getHeader: vi.fn((name: string) => name === 'host' ? 'user:password@remote.example:8443' : undefined),
+	};
+}
+
+describe('http-client-instrumentation', () => {
+	test('creates and completes a sanitized CLIENT span from diagnostics channels', () => {
+		const listeners = new Map<string, (message: unknown) => void>();
+		const span = {
+			end: vi.fn(),
+			recordException: vi.fn(),
+			setAttribute: vi.fn(),
+			setStatus: vi.fn(),
+		};
+		const tracer = { startSpan: vi.fn(() => span) } as any;
+		const unsubscribe = createHttpClientInstrumentation({
+			tracer,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+			subscribe: (name, listener) => {
+				listeners.set(name, listener);
+				return () => listeners.delete(name);
+			},
+		});
+		const clientRequest = request();
+
+		listeners.get('http.client.request.created')!({ request: clientRequest });
+		listeners.get('http.client.response.finish')!({
+			request: clientRequest,
+			response: { statusCode: 201, httpVersion: '1.1' },
+		});
+
+		expect(tracer.startSpan).toHaveBeenCalledWith('POST', {
+			kind: SpanKind.CLIENT,
+			attributes: {
+				'http.request.method': 'POST',
+				'url.full': 'https://remote.example:8443/inbox',
+				'server.address': 'remote.example',
+				'server.port': 8443,
+			},
+		});
+		expect(span.setAttribute).toHaveBeenCalledWith('http.response.status_code', 201);
+		expect(span.setAttribute).toHaveBeenCalledWith('network.protocol.version', '1.1');
+		expect(span.end).toHaveBeenCalledTimes(1);
+		unsubscribe();
+		expect(listeners).toHaveLength(0);
+	});
+
+	test('records a request error and ends the span once', () => {
+		const listeners = new Map<string, (message: unknown) => void>();
+		const span = { end: vi.fn(), recordException: vi.fn(), setAttribute: vi.fn(), setStatus: vi.fn() };
+		const error = Object.assign(new Error('connection refused'), { code: 'ECONNREFUSED' });
+		const clientRequest = request();
+		createHttpClientInstrumentation({
+			tracer: { startSpan: vi.fn(() => span) } as any,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+			subscribe: (name, listener) => {
+				listeners.set(name, listener);
+				return () => listeners.delete(name);
+			},
+		});
+
+		listeners.get('http.client.request.created')!({ request: clientRequest });
+		listeners.get('http.client.request.error')!({ request: clientRequest, error });
+		listeners.get('http.client.response.finish')!({ request: clientRequest, response: { statusCode: 200 } });
+
+		expect(span.recordException).toHaveBeenCalledWith(error);
+		expect(span.setAttribute).toHaveBeenCalledWith('error.type', 'ECONNREFUSED');
+		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR });
+		expect(span.end).toHaveBeenCalledTimes(1);
+	});
+
+	test('records the response status code as error.type for an error response', () => {
+		const listeners = new Map<string, (message: unknown) => void>();
+		const span = { end: vi.fn(), recordException: vi.fn(), setAttribute: vi.fn(), setStatus: vi.fn() };
+		const clientRequest = request();
+		createHttpClientInstrumentation({
+			tracer: { startSpan: vi.fn(() => span) } as any,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+			subscribe: (name, listener) => {
+				listeners.set(name, listener);
+				return () => listeners.delete(name);
+			},
+		});
+
+		listeners.get('http.client.request.created')!({ request: clientRequest });
+		listeners.get('http.client.response.finish')!({ request: clientRequest, response: { statusCode: 502 } });
+
+		expect(span.setAttribute).toHaveBeenCalledWith('error.type', '502');
+		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR });
+	});
+});

packages/backend/test/unit/core/telemetry/queue-instrumentation.ts

@@ -0,0 +1,57 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { beforeEach, describe, expect, test, vi } from 'vitest';
+import type * as Bull from 'bullmq';
+import { instrumentQueue } from '@/core/telemetry/queue-instrumentation.js';
+
+const mocks = vi.hoisted(() => ({
+	injectTraceContext: vi.fn((carrier: Record<string, string>) => {
+		carrier['traceparent'] = '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01';
+	}),
+}));
+
+vi.mock('@/core/telemetry/telemetry-registry.js', () => ({
+	injectTraceContext: mocks.injectTraceContext,
+}));
+
+describe('queue-instrumentation', () => {
+	beforeEach(() => {
+		mocks.injectTraceContext.mockClear();
+	});
+
+	test('injects the active trace context for add()', () => {
+		const add = vi.fn();
+		const queue = instrumentQueue({ add, addBulk: vi.fn() } as unknown as Bull.Queue<{ noteId: string }>);
+		const data = { noteId: '9d6b9a65-46c9-4e1b-a640-9589693893c9' };
+
+		queue.add('endedPollNotification', data);
+
+		expect(mocks.injectTraceContext).toHaveBeenCalledTimes(1);
+		expect(data).toMatchObject({
+			__misskeyTraceContext: {
+				traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01',
+			},
+		});
+		expect(add).toHaveBeenCalledWith('endedPollNotification', data, undefined);
+	});
+
+	test('injects every job passed to addBulk()', () => {
+		const addBulk = vi.fn();
+		const queue = instrumentQueue({ add: vi.fn(), addBulk } as unknown as Bull.Queue<{ to: string }>);
+		const jobs = [
+			{ name: 'deliver', data: { to: 'https://remote.example/inbox' } },
+			{ name: 'deliver', data: { to: 'https://remote2.example/inbox' } },
+		];
+
+		queue.addBulk(jobs);
+
+		expect(mocks.injectTraceContext).toHaveBeenCalledTimes(2);
+		expect(jobs).toEqual(expect.arrayContaining([
+			expect.objectContaining({ data: expect.objectContaining({ __misskeyTraceContext: expect.any(Object) }) }),
+		]));
+		expect(addBulk).toHaveBeenCalledWith(jobs);
+	});
+});

packages/backend/test/unit/core/telemetry/queue-trace-context.ts

@@ -0,0 +1,132 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { describe, expect, test, vi } from 'vitest';
+import type { Context, SpanContext } from '@opentelemetry/api';
+import { getQueueSpanContext, getQueueTraceContextMode, injectActiveTraceContext, injectQueueTraceContext } from '@/core/telemetry/queue-trace-context.js';
+
+const rootContext = {} as Context;
+const extractedContext = {} as Context;
+const sourceSpanContext: SpanContext = {
+	traceId: '0123456789abcdef0123456789abcdef',
+	spanId: '0123456789abcdef',
+	traceFlags: 1,
+	isRemote: true,
+};
+
+function jobData() {
+	return {
+		name: 'deliver',
+		__misskeyTraceContext: {
+			traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01',
+		},
+	};
+}
+
+describe('queue-trace-context', () => {
+	test('stores only a non-empty carrier in the job data', () => {
+		const data = { noteId: '9d6b9a65-46c9-4e1b-a640-9589693893c9' };
+
+		injectQueueTraceContext(data, carrier => {
+			carrier['traceparent'] = '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01';
+		});
+
+		expect(data).toMatchObject({
+			__misskeyTraceContext: {
+				traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01',
+			},
+		});
+	});
+
+	test('does not store an empty carrier when no active trace exists', () => {
+		const data = { noteId: '9d6b9a65-46c9-4e1b-a640-9589693893c9' };
+
+		injectQueueTraceContext(data, () => {});
+
+		expect(data).not.toHaveProperty('__misskeyTraceContext');
+	});
+
+	test('ignores non-object job data', () => {
+		const inject = vi.fn();
+
+		injectQueueTraceContext(null, inject);
+		injectQueueTraceContext('not a job object', inject);
+
+		expect(inject).not.toHaveBeenCalled();
+	});
+
+	test('injects the active context with the configured propagator', () => {
+		const activeContext = {} as Context;
+		const carrier = {};
+		const inject = vi.fn();
+
+		injectActiveTraceContext({
+			tracer: { startActiveSpan: vi.fn() } as any,
+			propagation: { inject, extract: vi.fn() } as any,
+			trace: { getSpanContext: vi.fn() },
+			getActiveContext: () => activeContext,
+			rootContext,
+			mode: 'link',
+			spanStatusCodeError: 2 as any,
+		}, carrier);
+
+		expect(inject).toHaveBeenCalledWith(activeContext, carrier);
+	});
+
+	test('starts a new root trace with a link by default', () => {
+		const extract = vi.fn(() => extractedContext);
+		const getSpanContext = vi.fn(() => sourceSpanContext);
+
+		const result = getQueueSpanContext(jobData(), {
+			rootContext,
+			propagation: { inject: vi.fn(), extract },
+			trace: { getSpanContext },
+			mode: 'link',
+		});
+
+		expect(extract).toHaveBeenCalledWith(rootContext, jobData().__misskeyTraceContext);
+		expect(result).toEqual({
+			options: {
+				root: true,
+				links: [{ context: sourceSpanContext }],
+			},
+			parentContext: rootContext,
+		});
+	});
+
+	test('uses the extracted context as the parent when parent mode is selected', () => {
+		const result = getQueueSpanContext(jobData(), {
+			rootContext,
+			propagation: { inject: vi.fn(), extract: () => extractedContext },
+			trace: { getSpanContext: () => sourceSpanContext },
+			mode: 'parent',
+		});
+
+		expect(result).toEqual({
+			options: {},
+			parentContext: extractedContext,
+		});
+	});
+
+	test('ignores malformed or missing carriers', () => {
+		const extract = vi.fn(() => extractedContext);
+		const deps = {
+			rootContext,
+			propagation: { inject: vi.fn(), extract },
+			trace: { getSpanContext: () => sourceSpanContext },
+			mode: 'link' as const,
+		};
+
+		expect(getQueueSpanContext({}, deps)).toBeUndefined();
+		expect(getQueueSpanContext({ __misskeyTraceContext: { traceparent: 1 } }, deps)).toBeUndefined();
+		expect(extract).not.toHaveBeenCalled();
+	});
+
+	test('defaults to link mode and rejects invalid configuration', () => {
+		expect(getQueueTraceContextMode(undefined)).toBe('link');
+		expect(getQueueTraceContextMode('parent')).toBe('parent');
+		expect(() => getQueueTraceContextMode('children')).toThrow('otelForBackend.jobTraceContextMode');
+	});
+});

packages/misskey-js/package.json

@@ -1,7 +1,7 @@
 {
 	"type": "module",
 	"name": "misskey-js",
-	"version": "2026.7.0-alpha.3",
+	"version": "2026.7.0-alpha.4",
 	"description": "Misskey SDK for JavaScript",
 	"license": "MIT",
 	"main": "./built/index.js",