diff --git a/extensions/diagnostics-otel/src/service.test.ts b/extensions/diagnostics-otel/src/service.test.ts index 1201995225eb..eddf26594a43 100644 --- a/extensions/diagnostics-otel/src/service.test.ts +++ b/extensions/diagnostics-otel/src/service.test.ts @@ -1027,6 +1027,27 @@ describe("diagnostics-otel service", () => { expect(telemetryState.tracer.startSpan).not.toHaveBeenCalled(); }); + test("attempts every provider shutdown and reports every failure", async () => { + const logError = new Error("log provider failed"); + const sdkError = new Error("SDK providers failed"); + logShutdown.mockRejectedValueOnce(logError); + sdkShutdown.mockRejectedValueOnce(sdkError); + const { service, ctx } = await startOtelService({ traces: true, metrics: true, logs: true }); + + const stopError = await Promise.resolve(service.stop?.(ctx)).catch((error: unknown) => error); + + expect(logShutdown).toHaveBeenCalledTimes(1); + expect(sdkShutdown).toHaveBeenCalledTimes(1); + expect(stopError).toBeInstanceOf(AggregateError); + expect(stopError).toMatchObject({ + errors: [logError, sdkError], + message: expect.stringContaining("log provider failed"), + }); + expect(stopError).toMatchObject({ + message: expect.stringContaining("SDK providers failed"), + }); + }); + test("registers and removes an OTLP exporter unhandled rejection handler", async () => { const { service, ctx } = await startOtelService({ traces: true, metrics: true, logs: true }); @@ -1058,6 +1079,38 @@ describe("diagnostics-otel service", () => { expect(unhandledRejectionHandlerState.getHandlers()).toHaveLength(0); }); + test("cleans up existing providers and does not reinitialize without capability", async () => { + const service = createDiagnosticsOtelService(); + const enabledCtx = createOtelContext(OTEL_TEST_ENDPOINT, { + traces: true, + metrics: true, + logs: true, + }); + await service.start(enabledCtx); + + sdkCtor.mockClear(); + sdkStart.mockClear(); + logExporterCtor.mockClear(); + const deniedCtx = createOtelContext(OTEL_TEST_ENDPOINT, { + traces: true, + metrics: true, + logs: true, + }); + delete deniedCtx.internalDiagnostics; + + await service.start(deniedCtx); + await service.stop?.(deniedCtx); + + expect(deniedCtx.logger.error).toHaveBeenCalledWith( + "diagnostics-otel: internal diagnostics capability unavailable", + ); + expect(sdkCtor).not.toHaveBeenCalled(); + expect(sdkStart).not.toHaveBeenCalled(); + expect(logExporterCtor).not.toHaveBeenCalled(); + expect(sdkShutdown).toHaveBeenCalledOnce(); + expect(logShutdown).toHaveBeenCalledOnce(); + }); + test("does not retain an OTLP exporter handler when startup setup fails", async () => { const startupError = new Error("trace exporter setup failed"); traceExporterCtor.mockImplementationOnce(() => { diff --git a/extensions/diagnostics-otel/src/service.ts b/extensions/diagnostics-otel/src/service.ts index 68c1f2c9f9e1..d80ec8d10e58 100644 --- a/extensions/diagnostics-otel/src/service.ts +++ b/extensions/diagnostics-otel/src/service.ts @@ -67,14 +67,28 @@ export function createDiagnosticsOtelService(): OpenClawPluginService { stopActiveTrustedSpans = null; unregisterUnhandledRejectionHandler = null; - currentUnregisterUnhandledRejectionHandler?.(); - currentUnsubscribe?.(); - currentStopActiveTrustedSpans?.(); - if (currentLogProvider) { - await currentLogProvider.shutdown().catch(() => undefined); + const settle = async (...stops: Array<(() => void | Promise) | null>) => + ( + await Promise.allSettled(stops.map((stop) => Promise.resolve().then(() => stop?.()))) + ).flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); + // Preserve cleanup -> provider flush -> handler removal while attempting every step per phase. + const failures = await settle(currentUnsubscribe, currentStopActiveTrustedSpans); + failures.push( + ...(await settle( + currentLogProvider ? () => currentLogProvider.shutdown() : null, + currentSdk ? () => currentSdk.shutdown() : null, + )), + ...(await settle(currentUnregisterUnhandledRejectionHandler)), + ); + + if (failures.length === 1) { + throw failures[0]; } - if (currentSdk) { - await currentSdk.shutdown().catch(() => undefined); + if (failures.length > 1) { + throw new AggregateError( + failures, + `diagnostics-otel shutdown failed: ${failures.join("; ")}`, + ); } }; @@ -128,6 +142,12 @@ export function createDiagnosticsOtelService(): OpenClawPluginService { if (enabledSignals.length === 0) { return; } + // This capability is the admission gate; no exporter may outlive a denied start. + const subscribe = ctx.internalDiagnostics?.onEvent; + if (!subscribe) { + ctx.logger.error("diagnostics-otel: internal diagnostics capability unavailable"); + return; + } const envProtocol = process.env.OTEL_EXPORTER_OTLP_PROTOCOL; const protocol = otel.protocol ?? (envProtocol?.trim() ? envProtocol : "http/protobuf"); @@ -308,11 +328,6 @@ export function createDiagnosticsOtelService(): OpenClawPluginService { ...createModelRecorders(recorderRuntime), ...createToolAndSystemRecorders(recorderRuntime), }; - const subscribe = ctx.internalDiagnostics?.onEvent; - if (!subscribe) { - ctx.logger.error("diagnostics-otel: internal diagnostics capability unavailable"); - return; - } unsubscribe = subscribe( createDiagnosticsEventHandler({ diff --git a/src/infra/diagnostic-events.test.ts b/src/infra/diagnostic-events.test.ts index 9f30bc270514..ed196aac7146 100644 --- a/src/infra/diagnostic-events.test.ts +++ b/src/infra/diagnostic-events.test.ts @@ -630,6 +630,46 @@ describe("diagnostic-events", () => { expect(events).toHaveLength(250); }); + it("does not extend a drain barrier for events queued after it starts", async () => { + const callIds: string[] = []; + onDiagnosticEvent((event) => { + if (event.type === "model.call.started") { + callIds.push(event.callId); + } + }); + + emitDiagnosticEvent({ + type: "model.call.started", + runId: "run-before-barrier", + callId: "before-barrier", + provider: "openai", + model: "gpt-5.4", + }); + const drained = waitForDiagnosticEventsDrained(); + for (let index = 0; index < 250; index += 1) { + emitDiagnosticEvent({ + type: "model.call.started", + runId: `run-after-${index}`, + callId: `after-${index}`, + provider: "openai", + model: "gpt-5.4", + }); + } + + await drained; + + expect(callIds).toHaveLength(100); + expect(callIds[0]).toBe("before-barrier"); + expect( + hasPendingInternalDiagnosticEvent( + (event) => event.type === "model.call.started" && event.callId === "after-249", + ), + ).toBe(true); + + await waitForDiagnosticEventsDrained(); + expect(callIds).toHaveLength(251); + }); + it("reports pending async diagnostic events before they drain", async () => { emitTrustedDiagnosticEvent({ type: "tool.execution.error", diff --git a/src/infra/diagnostic-events.ts b/src/infra/diagnostic-events.ts index e1667ebc72c5..c2a2d055b8d2 100644 --- a/src/infra/diagnostic-events.ts +++ b/src/infra/diagnostic-events.ts @@ -1216,10 +1216,15 @@ function dispatchAsyncDiagnosticDropSummary(state: DiagnosticEventsGlobalState): dispatchDiagnosticEvent(state, event, createInternalDiagnosticMetadata(false)); } -/** Waits until queued async diagnostic events have been delivered to listeners. */ +/** Waits until async diagnostic events queued when called are no longer pending. */ export async function waitForDiagnosticEventsDrained(): Promise { const state = getDiagnosticEventsState(); - while (state.asyncDrainScheduled || state.asyncQueue.length > 0) { + const targetSeq = state.asyncQueue.at(-1)?.event.seq; + if (targetSeq === undefined) { + return; + } + // The queue is append-ordered by seq, so a newer head means this snapshot drained or dropped. + while ((state.asyncQueue[0]?.event.seq ?? Number.POSITIVE_INFINITY) <= targetSeq) { await new Promise((resolve) => { setImmediate(resolve); }); diff --git a/src/plugins/services.test.ts b/src/plugins/services.test.ts index afa05af63c90..40f2bf940f9d 100644 --- a/src/plugins/services.test.ts +++ b/src/plugins/services.test.ts @@ -17,6 +17,11 @@ vi.mock("../logging/subsystem.js", () => ({ })); import { STATE_DIR } from "../config/paths.js"; +import { + emitTrustedDiagnosticEvent, + resetDiagnosticEventsForTest, + waitForDiagnosticEventsDrained, +} from "../infra/diagnostic-events.js"; import { queuePluginSessionsChanged, subscribePluginSessionsChanged } from "./gateway-events.js"; import { registerPluginHttpRoute } from "./http-registry.js"; import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "./runtime.js"; @@ -142,6 +147,7 @@ function createTrackingService( describe("startPluginServices", () => { beforeEach(() => { vi.clearAllMocks(); + resetDiagnosticEventsForTest(); resetPluginRuntimeStateForTest(); }); @@ -165,6 +171,82 @@ describe("startPluginServices", () => { expectServiceLifecycleState({ starts, stops, contexts, config }); }); + it("drains producer diagnostics before exporters stop and propagates exporter failures", async () => { + const order: string[] = []; + const producerError = new Error("producer stop failed"); + const exporterError = new Error("exporter stop failed"); + let unsubscribe: () => void = () => undefined; + const registry = createRegistry( + [ + { + id: "producer", + start: () => undefined, + stop: () => { + order.push("producer"); + emitTrustedDiagnosticEvent({ + type: "log.record", + level: "INFO", + message: "queued during producer shutdown", + }); + throw producerError; + }, + }, + ], + "plugin:test", + "workspace", + ); + registry.services.push( + ...createRegistry( + [ + { + id: "diagnostics-prometheus", + start: () => undefined, + stop: () => { + order.push("prometheus"); + }, + }, + ], + "diagnostics-prometheus", + "bundled", + ).services, + ...createRegistry( + [ + { + id: "diagnostics-otel", + start: (ctx) => { + unsubscribe = ctx.internalDiagnostics!.onEvent((event) => { + if (event.type === "log.record") { + order.push("event"); + } + }); + }, + stop: () => { + order.push("otel"); + unsubscribe(); + throw exporterError; + }, + }, + ], + "diagnostics-otel", + "bundled", + ).services, + ); + + const handle = await startPluginServices({ + registry, + config: createServiceConfig(), + }); + + await expect(handle.stop()).rejects.toBe(exporterError); + await waitForDiagnosticEventsDrained(); + + expect(order).toEqual(["producer", "event", "otel", "prometheus"]); + expect(mockedLogger.warn.mock.calls).toEqual([ + ["plugin service stop failed (producer): Error: producer stop failed"], + ["plugin service stop failed (diagnostics-otel): Error: exporter stop failed"], + ]); + }); + it("rolls back partially started services before starting their siblings", async () => { const acquired = new Set(); const received = vi.fn(); @@ -474,10 +556,15 @@ describe("startPluginServices", () => { await handle.stop(); }); - it("logs start/stop failures and continues", async () => { + it("attempts every ordinary service stop and preserves warn-and-continue failures", async () => { const stopOk = vi.fn(); - const stopThrows = vi.fn(() => { - throw new Error("stop failed"); + const firstError = new Error("first stop failed"); + const secondError = new Error("second stop failed"); + const stopFirst = vi.fn(() => { + throw firstError; + }); + const stopSecond = vi.fn(() => { + throw secondError; }); const handle = await startTrackingServices({ @@ -486,12 +573,13 @@ describe("startPluginServices", () => { failOnStart: true, stopSpy: vi.fn(), }), + createTrackingService("service-stop-first", { stopSpy: stopFirst }), createTrackingService("service-ok", { stopSpy: stopOk }), - createTrackingService("service-stop-fail", { stopSpy: stopThrows }), + createTrackingService("service-stop-second", { stopSpy: stopSecond }), ], }); - await handle.stop(); + await expect(handle.stop()).resolves.toBeUndefined(); expect(mockedLogger.error.mock.calls).toEqual([ [ @@ -500,10 +588,12 @@ describe("startPluginServices", () => { ]); expect(requireLoggerErrorMessage()).not.toContain("\n"); expect(mockedLogger.warn.mock.calls).toEqual([ - ["plugin service stop failed (service-stop-fail): Error: stop failed"], + ["plugin service stop failed (service-stop-second): Error: second stop failed"], + ["plugin service stop failed (service-stop-first): Error: first stop failed"], ]); expect(stopOk).toHaveBeenCalledOnce(); - expect(stopThrows).toHaveBeenCalledOnce(); + expect(stopFirst).toHaveBeenCalledOnce(); + expect(stopSecond).toHaveBeenCalledOnce(); }); it("continues starting siblings when rollback also fails", async () => { diff --git a/src/plugins/services.ts b/src/plugins/services.ts index ead9ac142b04..3ac4cade3f3c 100644 --- a/src/plugins/services.ts +++ b/src/plugins/services.ts @@ -5,6 +5,7 @@ import type { GatewayPluginEventBroadcastFn } from "../gateway/server-broadcast- import { emitTrustedDiagnosticEventWithPrivateData, onTrustedInternalDiagnosticEvent, + waitForDiagnosticEventsDrained, } from "../infra/diagnostic-events.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { subscribePluginSessionsChanged } from "./gateway-events.js"; @@ -171,16 +172,18 @@ export async function startPluginServices(params: { }): Promise { const running: Array<{ id: string; + diagnosticsExporter: boolean; stop?: () => void | Promise; revokeGatewayEvents: () => void; }> = []; - const stopService = async (entry: (typeof running)[number]) => { + const stopService = async (entry: (typeof running)[number], failures?: unknown[]) => { try { if (entry.stop) { await withPluginHttpRouteRegistry(params.registry, () => entry.stop?.()); } } catch (err) { log.warn(`plugin service stop failed (${entry.id}): ${String(err)}`); + failures?.push(err); } finally { entry.revokeGatewayEvents(); } @@ -202,6 +205,7 @@ export async function startPluginServices(params: { }); const runningService = { id: service.id, + diagnosticsExporter: serviceContext.internalDiagnostics !== undefined, stop: service.stop ? () => service.stop?.(serviceContext) : undefined, revokeGatewayEvents: scopedGatewayEvents.revoke, }; @@ -235,8 +239,30 @@ export async function startPluginServices(params: { stop: () => // Store the shared promise before plugin cleanup runs so shutdown cannot start twice. (stopPromise ??= Promise.resolve().then(async () => { - for (const entry of running.toReversed()) { - await stopService(entry); + const reversed = running.toReversed(); + const diagnosticsExporters = reversed.filter((entry) => entry.diagnosticsExporter); + const exporterFailures: unknown[] = []; + const stopServices = async (services: typeof reversed, failures?: unknown[]) => { + for (const entry of services) { + await stopService(entry, failures); + } + }; + await stopServices(reversed.filter((entry) => !entry.diagnosticsExporter)); + if (diagnosticsExporters.length > 0) { + // Producers stop first; this barrier preserves their queued tail before exporters detach. + await waitForDiagnosticEventsDrained(); + } + // Ordinary plugin cleanup stays warn-and-continue. Trusted diagnostics + // exporter failures propagate because they can mean telemetry was lost. + await stopServices(diagnosticsExporters, exporterFailures); + if (exporterFailures.length === 1) { + throw exporterFailures[0]; + } + if (exporterFailures.length > 1) { + throw new AggregateError( + exporterFailures, + "multiple diagnostics exporters failed to stop", + ); } })), };