fix(backend): address OpenTelemetry review feedback

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

@@ -50,6 +50,8 @@ type CreateResourceDeps = {
 	serviceVersion: string;
 };
 
+type InstrumentationInstaller = () => (() => void) | Promise<() => void>;
+
 export class OpenTelemetryAdapter implements TelemetryAdapter {
 	public constructor(
 		private readonly deps: OpenTelemetryAdapterDeps,
@@ -111,39 +113,46 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 
 		// provider操作をdepsに閉じ込め、span wrapper本体をユニットテストしやすくする。
 		const tracer = provider.getTracer('misskey-backend');
-
-		return new OpenTelemetryAdapter({
-			tracer,
-			provider,
-			getActiveSpan: () => trace.getActiveSpan(),
-			spanStatusCodeError: SpanStatusCode.ERROR,
-			shutdownTimeout: DEFAULT_SHUTDOWN_TIMEOUT,
-			shutdownHttpClientInstrumentation: installHttpClientInstrumentation({
+		return installInstrumentationsWithCleanup([
+			() => installHttpClientInstrumentation({
 				tracer,
 				spanKindClient: SpanKind.CLIENT,
 				spanStatusCodeError: SpanStatusCode.ERROR,
 			}),
 			// pg のrequire hookとioredis diagnostics channelは、Nest moduleの動的importより前に有効化する。
-			shutdownDatabaseInstrumentation: await installDatabaseInstrumentation(provider, {
+			() => installDatabaseInstrumentation(provider, {
 				capturePgSpans: config.capturePgSpans === true,
 				capturePgStatement: config.capturePgStatement === true,
 				capturePgConnectionSpans: config.capturePgConnectionSpans === true,
 			}),
-			shutdownRedisInstrumentation: installRedisInstrumentation(tracer, SpanKind.CLIENT, SpanStatusCode.ERROR, {
+			() => installRedisInstrumentation(tracer, SpanKind.CLIENT, SpanStatusCode.ERROR, {
 				captureConnectionSpans: config.captureRedisConnectionSpans === true,
 				captureCommandSpans: config.captureRedisCommandSpans === true,
 				requireParentSpan: config.captureRedisRootSpans !== true,
 			}),
-			queueTraceContext: {
+		], ([
+			shutdownHttpClientInstrumentation,
+			shutdownDatabaseInstrumentation,
+			shutdownRedisInstrumentation,
+		]) => new OpenTelemetryAdapter({
 				tracer,
-				propagation,
-				trace,
-				getActiveContext: () => context.active(),
-				rootContext: ROOT_CONTEXT,
-				mode: getQueueTraceContextMode(config.jobTraceContextMode),
+				provider,
+				getActiveSpan: () => trace.getActiveSpan(),
 				spanStatusCodeError: SpanStatusCode.ERROR,
-			},
-		});
+				shutdownTimeout: DEFAULT_SHUTDOWN_TIMEOUT,
+				shutdownHttpClientInstrumentation,
+				shutdownDatabaseInstrumentation,
+				shutdownRedisInstrumentation,
+				queueTraceContext: {
+					tracer,
+					propagation,
+					trace,
+					getActiveContext: () => context.active(),
+					rootContext: ROOT_CONTEXT,
+					mode: getQueueTraceContextMode(config.jobTraceContextMode),
+					spanStatusCodeError: SpanStatusCode.ERROR,
+				},
+			}));
 	}
 
 	public captureMessage(message: string, _opts: TelemetryCaptureMessageOptions): void {
@@ -192,9 +201,11 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 	}
 
 	public async shutdown(): Promise<void> {
-		this.deps.shutdownHttpClientInstrumentation?.();
-		this.deps.shutdownDatabaseInstrumentation?.();
-		this.deps.shutdownRedisInstrumentation?.();
+		shutdownInstrumentations([
+			this.deps.shutdownHttpClientInstrumentation,
+			this.deps.shutdownDatabaseInstrumentation,
+			this.deps.shutdownRedisInstrumentation,
+		]);
 		// BatchSpanProcessorのflushが詰まってもプロセス終了を妨げないよう、上限時間を設ける。
 		// タイムアウト側のtimerは、flushが先に終わった場合にイベントループを無駄に引き留めないようclearする。
 		let timer: NodeJS.Timeout | undefined;
@@ -209,6 +220,32 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 	}
 }
 
+export async function installInstrumentationsWithCleanup<T>(
+	installers: readonly InstrumentationInstaller[],
+	create: (shutdowns: ReadonlyArray<() => void>) => T,
+): Promise<T> {
+	const shutdowns: Array<() => void> = [];
+	try {
+		for (const install of installers) {
+			shutdowns.push(await install());
+		}
+		return create(shutdowns);
+	} catch (error) {
+		shutdownInstrumentations(shutdowns.reverse());
+		throw error;
+	}
+}
+
+function shutdownInstrumentations(shutdowns: ReadonlyArray<(() => void) | undefined>): void {
+	for (const shutdown of shutdowns) {
+		try {
+			shutdown?.();
+		} catch {
+			// instrumentation の停止失敗で、残りの停止処理や provider の flush を妨げない。
+		}
+	}
+}
+
 export function createResource(config: OtelBackendRuntimeConfig, deps: CreateResourceDeps): Resource {
 	// resourceを明示指定するとSDKのdefaultResource()は自動付与されなくなる(マージではなく上書き)ため、
 	// telemetry.sdk.*等の標準属性を失わないよう明示的にmergeする。

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

@@ -3,13 +3,13 @@
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
-import { beforeEach, describe, expect, test, vi } from 'vitest';
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
 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';
+import { OpenTelemetryAdapter, createResource, createSampler, getMisskeyProcessRole, installInstrumentationsWithCleanup } from '@/core/telemetry/adapters/OpenTelemetryAdapter.js';
 
 const mocks = vi.hoisted(() => {
 	return {
@@ -40,6 +40,10 @@ const samplerDeps = {
 };
 
 describe('OpenTelemetryAdapter', () => {
+	afterEach(() => {
+		vi.useRealTimers();
+	});
+
 	test('wraps async work in an active span and ends it after success', async () => {
 		const span = {
 			end: vi.fn(),
@@ -216,7 +220,6 @@ describe('OpenTelemetryAdapter', () => {
 		await vi.advanceTimersByTimeAsync(50);
 
 		await expect(shutdown).resolves.toBeUndefined();
-		vi.useRealTimers();
 	});
 
 	test('clears the shutdown timeout timer once provider.shutdown() resolves first', async () => {
@@ -234,7 +237,30 @@ describe('OpenTelemetryAdapter', () => {
 
 		expect(clearTimeoutSpy).toHaveBeenCalled();
 		clearTimeoutSpy.mockRestore();
-		vi.useRealTimers();
+	});
+
+	test('continues shutdown and flushes the provider when instrumentation cleanup fails', async () => {
+		const shutdownHttpClientInstrumentation = vi.fn(() => { throw new Error('failed'); });
+		const shutdownDatabaseInstrumentation = vi.fn();
+		const shutdownRedisInstrumentation = vi.fn();
+		const provider = { shutdown: vi.fn().mockResolvedValue(undefined) };
+		const adapter = new OpenTelemetryAdapter({
+			tracer: { startActiveSpan: vi.fn() },
+			provider,
+			getActiveSpan: () => undefined,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+			shutdownTimeout: 10,
+			shutdownHttpClientInstrumentation,
+			shutdownDatabaseInstrumentation,
+			shutdownRedisInstrumentation,
+		});
+
+		await expect(adapter.shutdown()).resolves.toBeUndefined();
+
+		expect(shutdownHttpClientInstrumentation).toHaveBeenCalledOnce();
+		expect(shutdownDatabaseInstrumentation).toHaveBeenCalledOnce();
+		expect(shutdownRedisInstrumentation).toHaveBeenCalledOnce();
+		expect(provider.shutdown).toHaveBeenCalledOnce();
 	});
 
 	test('captureMessage starts a standalone span to report the error when there is no active span', () => {
@@ -271,6 +297,38 @@ describe('OpenTelemetryAdapter', () => {
 	});
 });
 
+describe('installInstrumentationsWithCleanup', () => {
+	test('cleans up acquired instrumentations in reverse order and preserves the installation error', async () => {
+		const installationError = new Error('failed to install');
+		const shutdownHttpClientInstrumentation = vi.fn();
+		const shutdownDatabaseInstrumentation = vi.fn(() => { throw new Error('failed to clean up'); });
+
+		await expect(installInstrumentationsWithCleanup([
+			() => shutdownHttpClientInstrumentation,
+			() => shutdownDatabaseInstrumentation,
+			() => { throw installationError; },
+		], () => { throw new Error('should not create'); })).rejects.toBe(installationError);
+
+		expect(shutdownDatabaseInstrumentation).toHaveBeenCalledOnce();
+		expect(shutdownHttpClientInstrumentation).toHaveBeenCalledOnce();
+		expect(shutdownDatabaseInstrumentation.mock.invocationCallOrder[0]).toBeLessThan(shutdownHttpClientInstrumentation.mock.invocationCallOrder[0]);
+	});
+
+	test('cleans up all instrumentations when adapter construction fails', async () => {
+		const constructionError = new Error('failed to create adapter');
+		const shutdownHttpClientInstrumentation = vi.fn();
+		const shutdownDatabaseInstrumentation = vi.fn();
+
+		await expect(installInstrumentationsWithCleanup([
+			() => shutdownHttpClientInstrumentation,
+			() => shutdownDatabaseInstrumentation,
+		], () => { throw constructionError; })).rejects.toBe(constructionError);
+
+		expect(shutdownDatabaseInstrumentation).toHaveBeenCalledOnce();
+		expect(shutdownHttpClientInstrumentation).toHaveBeenCalledOnce();
+	});
+});
+
 describe('createSampler', () => {
 	test('accepts sample rates within [0, 1]', () => {
 		expect(() => createSampler(0, samplerDeps)).not.toThrow();

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

@@ -4,8 +4,9 @@
  */
 
 import { describe, expect, test, vi } from 'vitest';
+import { SpanStatusCode } from '@opentelemetry/api';
 import type { Context, SpanContext } from '@opentelemetry/api';
-import { getQueueSpanContext, getQueueTraceContextMode, injectActiveTraceContext, injectQueueTraceContext } from '@/core/telemetry/queue-trace-context.js';
+import { executeSpan, getQueueSpanContext, getQueueTraceContextMode, injectActiveTraceContext, injectQueueTraceContext, recordSpanError, startSpanWithQueueTraceContext } from '@/core/telemetry/queue-trace-context.js';
 
 const rootContext = { kind: 'root' } as unknown as Context;
 const extractedContext = { kind: 'extracted' } as unknown as Context;
@@ -132,4 +133,88 @@ describe('queue-trace-context', () => {
 		expect(getQueueTraceContextMode('parent')).toBe('parent');
 		expect(() => getQueueTraceContextMode('children')).toThrow('otelForBackend.jobTraceContextMode');
 	});
+
+	test('executes synchronous work and ends the span once', () => {
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() };
+
+		expect(executeSpan(span as any, () => 'ok', SpanStatusCode.ERROR)).toBe('ok');
+		expect(span.end).toHaveBeenCalledOnce();
+		expect(span.recordException).not.toHaveBeenCalled();
+	});
+
+	test('waits for resolved Promise work before ending the span', async () => {
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() };
+
+		await expect(executeSpan(span as any, () => Promise.resolve('ok'), SpanStatusCode.ERROR)).resolves.toBe('ok');
+		expect(span.end).toHaveBeenCalledOnce();
+		expect(span.recordException).not.toHaveBeenCalled();
+	});
+
+	test('records and propagates synchronous failures', () => {
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() };
+		const error = new Error('boom');
+
+		expect(() => executeSpan(span as any, () => { throw error; }, SpanStatusCode.ERROR)).toThrow(error);
+		expect(span.recordException).toHaveBeenCalledWith(error);
+		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: error.message });
+		expect(span.end).toHaveBeenCalledOnce();
+	});
+
+	test('records and propagates rejected Promise work', async () => {
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() };
+		const error = new Error('boom');
+
+		await expect(executeSpan(span as any, () => Promise.reject(error), SpanStatusCode.ERROR)).rejects.toBe(error);
+		expect(span.recordException).toHaveBeenCalledWith(error);
+		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: error.message });
+		expect(span.end).toHaveBeenCalledOnce();
+	});
+
+	test('normalizes non-Error failures before recording them', () => {
+		const span = { recordException: vi.fn(), setStatus: vi.fn() };
+
+		recordSpanError(span as any, 'boom', SpanStatusCode.ERROR);
+
+		expect(span.recordException).toHaveBeenCalledWith(expect.objectContaining({ message: 'boom' }));
+		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: 'boom' });
+	});
+
+	test('starts a span with the extracted queue context', () => {
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() };
+		const startActiveSpan = vi.fn((_name: string, _options: unknown, _context: unknown, fn: (spanArg: typeof span) => string) => fn(span));
+		const deps = {
+			tracer: { startActiveSpan } as any,
+			rootContext,
+			propagation: { inject: vi.fn(), extract: () => extractedContext } as any,
+			trace: { getSpanContext: () => sourceSpanContext },
+			getActiveContext: () => rootContext,
+			mode: 'link' as const,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		};
+
+		expect(startSpanWithQueueTraceContext(deps, 'Queue: Deliver', jobData(), () => 'ok', () => 'fallback')).toBe('ok');
+		expect(startActiveSpan).toHaveBeenCalledWith('Queue: Deliver', {
+			root: true,
+			links: [{ context: sourceSpanContext }],
+		}, rootContext, expect.any(Function));
+		expect(span.end).toHaveBeenCalledOnce();
+	});
+
+	test('uses the fallback without starting a span when queue context is missing', () => {
+		const startActiveSpan = vi.fn();
+		const fallback = vi.fn(() => 'fallback');
+		const deps = {
+			tracer: { startActiveSpan } as any,
+			rootContext,
+			propagation: { inject: vi.fn(), extract: vi.fn() } as any,
+			trace: { getSpanContext: vi.fn() },
+			getActiveContext: () => rootContext,
+			mode: 'link' as const,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		};
+
+		expect(startSpanWithQueueTraceContext(deps, 'Queue: Deliver', {}, () => 'ok', fallback)).toBe('fallback');
+		expect(fallback).toHaveBeenCalledOnce();
+		expect(startActiveSpan).not.toHaveBeenCalled();
+	});
 });

packages/backend/test/unit/telemetry-database-instrumentation.ts

@@ -95,7 +95,7 @@ describe('database-instrumentation', () => {
 		}));
 	});
 
-	test('cleans up both instrumentations when initialization fails', () => {
+	test('cleans up the pg instrumentation when initialization fails', () => {
 		const pg = { setTracerProvider: vi.fn(), enable: vi.fn(), disable: vi.fn() };
 		pg.enable.mockImplementation(() => { throw new Error('failed'); });