feat: BullMQのジョブ追加時に発動する手動計測の追加 (#17710)

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

* add comment

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

@@ -8,10 +8,12 @@ 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;
 
@@ -24,6 +26,7 @@ type OpenTelemetryAdapterDeps = {
 	spanStatusCodeError: SpanStatusCode;
 	shutdownTimeout: number;
 	shutdownHttpClientInstrumentation?: () => void;
+	queueTraceContext?: QueueTraceContextDeps;
 };
 
 type CreateSamplerDeps = {
@@ -50,7 +53,7 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 
 	public static async create(config: OtelBackendRuntimeConfig): Promise<OpenTelemetryAdapter> {
 		const [
-			{ diag, DiagLogLevel, SpanKind, SpanStatusCode, trace },
+			{ context, diag, DiagLogLevel, propagation, ROOT_CONTEXT, SpanKind, SpanStatusCode, trace },
 			{ W3CTraceContextPropagator },
 			{ OTLPTraceExporter },
 			{ defaultResource, detectResources, envDetector, resourceFromAttributes },
@@ -115,6 +118,15 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 				spanKindClient: SpanKind.CLIENT,
 				spanStatusCodeError: SpanStatusCode.ERROR,
 			}),
+			queueTraceContext: {
+				tracer,
+				propagation,
+				trace,
+				getActiveContext: () => context.active(),
+				rootContext: ROOT_CONTEXT,
+				mode: getQueueTraceContextMode(config.jobTraceContextMode),
+				spanStatusCodeError: SpanStatusCode.ERROR,
+			},
 		});
 	}
 
@@ -124,45 +136,34 @@ 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> {
@@ -211,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/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()));