From 1f0147921bcb4dcc3fbf1f129be036b647d77bca Mon Sep 17 00:00:00 2001 From: Rihan Arfan Date: Thu, 9 Jul 2026 14:51:54 +0100 Subject: [PATCH 1/6] presets(cloudflare): bridge tracing channel events to observability custom spans --- docs/2.deploy/20.providers/cloudflare.md | 27 ++ src/presets/cloudflare/preset.ts | 26 +- .../cloudflare/runtime/telemetry/plugin.ts | 115 +++++++ test/unit/cloudflare-telemetry.test.ts | 291 ++++++++++++++++++ 4 files changed, 458 insertions(+), 1 deletion(-) create mode 100644 src/presets/cloudflare/runtime/telemetry/plugin.ts create mode 100644 test/unit/cloudflare-telemetry.test.ts diff --git a/docs/2.deploy/20.providers/cloudflare.md b/docs/2.deploy/20.providers/cloudflare.md index 1eb87cbae6..d58ab23e2e 100644 --- a/docs/2.deploy/20.providers/cloudflare.md +++ b/docs/2.deploy/20.providers/cloudflare.md @@ -103,6 +103,33 @@ export default defineConfig({ No manual Wrangler configuration is needed - Nitro handles it for you. +### Tracing + +**๐Ÿงช Experimental!** + +When the experimental [`tracingChannel`](/config#tracingchannel) option is enabled, the Cloudflare presets report Nitro's tracing-channel events (h3 routes and middleware, srvx, unstorage operations, โ€ฆ) as [custom spans](https://developers.cloudflare.com/workers/observability/traces/custom-spans/), alongside Cloudflare's automatic instrumentation (fetch calls, KV reads, D1 queries, โ€ฆ) โ€” no OpenTelemetry SDK required. + +```ts [nitro.config.ts] +import { defineConfig } from "nitro"; + +export default defineConfig({ + preset: "cloudflare_module", + tracingChannel: true, +}); +``` + +Tracing must be enabled on the Worker for spans to be recorded: + +```jsonc [wrangler.jsonc] +{ + "observability": { + "traces": { + "enabled": true + } + } +} +``` + ## Cloudflare Pages **Preset:** `cloudflare_pages` diff --git a/src/presets/cloudflare/preset.ts b/src/presets/cloudflare/preset.ts index 439011c4f2..2e94a85636 100644 --- a/src/presets/cloudflare/preset.ts +++ b/src/presets/cloudflare/preset.ts @@ -2,7 +2,8 @@ import { defineNitroPreset } from "../_utils/preset.ts"; import { writeFile } from "../_utils/fs.ts"; import type { Nitro } from "nitro/types"; import type { Plugin } from "rollup"; -import { resolve } from "pathe"; +import { join, resolve } from "pathe"; +import { presetsDir } from "nitro/meta"; import { unenvCfExternals } from "./unenv/preset.ts"; import { enableNodeCompat, @@ -141,6 +142,14 @@ export const cloudflareDev = defineNitroPreset( devServer: { runner: "miniflare", }, + hooks: { + "build:before": (nitro) => { + // The bridge imports `cloudflare:workers`, only available in workerd + if (nitro.options.devServer.runner === "miniflare") { + setupTracingBridge(nitro); + } + }, + }, }, { name: "cloudflare-dev" as const, @@ -180,6 +189,7 @@ const cloudflareModule = defineNitroPreset( nitro.options.unenv.push(unenvCfExternals); await enableNodeCompat(nitro); await setupEntryExports(nitro); + setupTracingBridge(nitro); }, async compiled(nitro: Nitro) { await writeWranglerConfig(nitro, "module"); @@ -219,3 +229,17 @@ export default [ cloudflareDurable, cloudflareDev, ]; + +/** + * Export tracing-channel spans as Cloudflare custom spans (`tracing.enterSpan`) + * Registered first (unshift) so the bridge subscribes to the traced channels at + * startup, before any request is handled. + */ +function setupTracingBridge(nitro: Nitro) { + if (!nitro.options.tracingChannel) { + return; + } + nitro.options.plugins ??= []; + + nitro.options.plugins.unshift(join(presetsDir, "cloudflare/runtime/telemetry/plugin")); +} diff --git a/src/presets/cloudflare/runtime/telemetry/plugin.ts b/src/presets/cloudflare/runtime/telemetry/plugin.ts new file mode 100644 index 0000000000..9dff993495 --- /dev/null +++ b/src/presets/cloudflare/runtime/telemetry/plugin.ts @@ -0,0 +1,115 @@ +import { definePlugin } from "nitro"; +// Handle older compatibility dates without the custom-spans API +import * as cloudflare from "cloudflare:workers"; +import type { Span, Tracing } from "@cloudflare/workers-types"; +import type { IAnyValue } from "#nitro/runtime/telemetry/types"; +import { TRACED_CHANNELS } from "#nitro/runtime/telemetry/channels"; + +// https://developers.cloudflare.com/workers/observability/traces/custom-spans/ +const tracing = (cloudflare as { tracing?: Tracing }).tracing; + +interface PendingSpan { + span: Span; + close: () => void; +} + +/** + * Exports Nitro tracing-channel events as Cloudflare Workers custom spans, + * alongside Cloudflare's automatic instrumentation (fetch, KV, D1, โ€ฆ) + */ +export default definePlugin(() => { + const diagnostics = globalThis.process?.getBuiltinModule?.("node:diagnostics_channel"); + if (!diagnostics?.subscribe || typeof tracing?.enterSpan !== "function") return; + + // Open span + closer per in-flight operation, keyed by the context object + // `tracingChannel` publishes to every phase of the same operation. + const pending = new WeakMap(); + + for (const name of Object.keys(TRACED_CHANNELS)) { + const describe = TRACED_CHANNELS[name]; + + diagnostics.subscribe(`tracing:${name}:start`, (message) => { + try { + // The span name is fixed at creation (Cloudflare has no rename API), + // so it is derived from the start payload; end-time data (status code, + // matched route) lands in the attributes set at `asyncEnd`. + const info = describe(name, message); + + let close!: () => void; + const done = new Promise((resolve) => { + close = resolve; + }); + tracing.enterSpan(info.name, (span) => { + pending.set(message as object, { span, close }); + return done; + }); + } catch { + // Malformed payload, or tracing rejected the span (e.g. outside a + // request) โ€” skip it, never break the traced operation. + } + }); + + const finalize = (message: unknown) => { + const entry = pending.get(message as object); + if (!entry) return; + pending.delete(message as object); + try { + // Skip attribute work for unsampled requests (`head_sampling_rate`). + if (entry.span.isTraced) { + // Re-describe on the completed payload: by now producers have set + // end-time fields (`result.status`, `context.matchedRoute`, โ€ฆ). + const info = describe(name, message); + for (const { key, value } of info.attributes) { + entry.span.setAttribute(key, attributeValue(value)); + } + const error = (message as { error?: unknown }).error; + if (error !== undefined) { + recordException(entry.span, error); + } + } + } catch { + // Never break the traced operation. + } finally { + entry.close(); + } + }; + + diagnostics.subscribe(`tracing:${name}:asyncEnd`, finalize); + + // `tracePromise` never publishes `asyncEnd` when the traced function + // throws synchronously โ€” only `end`, with `error` already set. In the + // normal async path `end` fires before the promise settles, while `error` + // is still unset, so the guard makes this a no-op there. + diagnostics.subscribe(`tracing:${name}:end`, (message) => { + if ((message as { error?: unknown }).error !== undefined) { + finalize(message); + } + }); + } +}); + +/** OTLP `IAnyValue` (from the shared describers) โ†’ Cloudflare attribute value. */ +function attributeValue(value: IAnyValue): string | number | boolean | undefined { + if (value.stringValue != null) return value.stringValue; + if (value.intValue != null) return value.intValue; + if (value.doubleValue != null) return value.doubleValue; + if (value.boolValue != null) return value.boolValue; +} + +/** + * OTEL exception semconv, flattened onto span attributes โ€” the Cloudflare API + * has no span events, and `setOutcome` is not available yet. + */ +function recordException(span: Span, error: unknown): void { + const err = error as Partial | undefined; + if (typeof err?.name === "string") { + span.setAttribute("exception.type", err.name); + } + span.setAttribute( + "exception.message", + typeof err?.message === "string" ? err.message : String(error) + ); + if (typeof err?.stack === "string") { + span.setAttribute("exception.stacktrace", err.stack); + } +} diff --git a/test/unit/cloudflare-telemetry.test.ts b/test/unit/cloudflare-telemetry.test.ts new file mode 100644 index 0000000000..03c1b5833b --- /dev/null +++ b/test/unit/cloudflare-telemetry.test.ts @@ -0,0 +1,291 @@ +import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import type { TracingChannel } from "node:diagnostics_channel"; + +// --------------------------------------------------------------------------- +// Cloudflare custom-spans simulation (`tracing.enterSpan`) +// +// Mirrors the documented behavior: the callback runs synchronously inside the +// span, and the span ends when the callback returns or its returned promise +// settles. https://developers.cloudflare.com/workers/observability/traces/custom-spans/ +// --------------------------------------------------------------------------- + +interface FakeSpan { + name: string; + attributes: Record; + isTraced: boolean; + ended: boolean; + setAttribute(key: string, value?: string | number | boolean): void; +} + +const harness = vi.hoisted(() => ({ + spans: [] as FakeSpan[], + // Sampling decision applied to newly created spans (`head_sampling_rate`). + isTraced: true, +})); + +vi.mock("nitro", () => ({ definePlugin: (def: unknown) => def })); + +vi.mock("cloudflare:workers", () => ({ + tracing: { + enterSpan(name: string, callback: (span: FakeSpan) => unknown) { + const span: FakeSpan = { + name, + attributes: {}, + isTraced: harness.isTraced, + ended: false, + setAttribute(key, value) { + if (this.ended) { + throw new Error(`setAttribute("${key}") after span ended`); + } + if (value !== undefined) { + this.attributes[key] = value; + } + }, + }; + harness.spans.push(span); + const result = callback(span); + if (result && typeof (result as Promise).then === "function") { + (result as Promise).then( + () => { + span.ended = true; + }, + () => { + span.ended = true; + } + ); + } else { + span.ended = true; + } + return result; + }, + }, +})); + +import telemetryPlugin from "../../src/presets/cloudflare/runtime/telemetry/plugin.ts"; + +const diagnostics = process.getBuiltinModule("node:diagnostics_channel"); + +// --------------------------------------------------------------------------- +// Test helpers +// --------------------------------------------------------------------------- + +/** Run one traced operation the way producers (h3/srvx/unstorage) do. */ +function traced( + channelName: string, + data: Record, + fn: () => Promise = async () => "ok" +): Promise { + const channel = diagnostics.tracingChannel(channelName) as TracingChannel< + Record + >; + return channel.tracePromise(fn, data); +} + +/** + * Wait for span closure: the deferred the plugin hands to `enterSpan` resolves + * at `asyncEnd`, and the simulated runtime observes it one microtask later. + */ +function settled(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +// --------------------------------------------------------------------------- +// Setup: activate the plugin once (it subscribes to the traced channels +// process-wide via `diagnostics_channel.subscribe`; no globals are patched). +// --------------------------------------------------------------------------- + +const originalTracingChannel = diagnostics.tracingChannel; + +beforeAll(() => { + (telemetryPlugin as unknown as () => void)(); +}); + +beforeEach(() => { + harness.spans.length = 0; + harness.isTraced = true; +}); + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe("cloudflare telemetry span lifecycle", () => { + it("opens a span at operation start and closes it at asyncEnd", async () => { + let midOperation: FakeSpan | undefined; + await traced("unstorage.setItem", {}, async () => { + // The span exists (and is still open) while the operation runs. + midOperation = harness.spans[0]; + expect(midOperation).toBeDefined(); + expect(midOperation!.ended).toBe(false); + return "ok"; + }); + await settled(); + + expect(harness.spans).toHaveLength(1); + expect(harness.spans[0]).toBe(midOperation); + expect(harness.spans[0].name).toBe("setItem"); + expect(harness.spans[0].ended).toBe(true); + expect(harness.spans[0].attributes["db.operation"]).toBe("setItem"); + }); + + it("closes overlapping operations independently", async () => { + let releaseFirst!: () => void; + const firstGate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const first = traced("unstorage.getItem", { base: "cache" }, () => firstGate.then(() => "a")); + const second = traced("unstorage.setItem", {}, async () => "b"); + + await second; + await settled(); + expect(harness.spans).toHaveLength(2); + const [firstSpan, secondSpan] = harness.spans; + expect(secondSpan.ended).toBe(true); + expect(firstSpan.ended).toBe(false); + + releaseFirst(); + await first; + await settled(); + expect(firstSpan.ended).toBe(true); + expect(firstSpan.name).toBe("getItem cache"); + }); + + it("still closes the span when the operation rejects", async () => { + const error = new Error("async failure"); + await expect( + traced("unstorage.setItem", {}, async () => { + throw error; + }) + ).rejects.toThrow("async failure"); + await settled(); + + const [span] = harness.spans; + expect(span.ended).toBe(true); + expect(span.attributes["exception.type"]).toBe("Error"); + expect(span.attributes["exception.message"]).toBe("async failure"); + expect(span.attributes["exception.stacktrace"]).toBe(error.stack); + }); + + it("closes the span when the traced function throws synchronously", async () => { + // A sync throw makes `tracePromise` publish `end` (with `error` set) and + // rethrow without ever publishing `asyncEnd`. + const error = new Error("sync failure"); + expect(() => + traced("unstorage.setItem", {}, () => { + throw error; + }) + ).toThrow("sync failure"); + await settled(); + + const [span] = harness.spans; + expect(span.ended).toBe(true); + expect(span.attributes["exception.type"]).toBe("Error"); + expect(span.attributes["exception.message"]).toBe("sync failure"); + }); + + it("records non-Error rejections as exception.message only", async () => { + await expect( + traced("unstorage.setItem", {}, async () => { + throw "plain failure"; // eslint-disable-line no-throw-literal + }) + ).rejects.toBe("plain failure"); + await settled(); + + const [span] = harness.spans; + expect(span.attributes["exception.message"]).toBe("plain failure"); + expect(span.attributes["exception.type"]).toBeUndefined(); + expect(span.attributes["exception.stacktrace"]).toBeUndefined(); + }); + + it("skips attribute work for unsampled requests but still closes the span", async () => { + harness.isTraced = false; + await traced("unstorage.setItem", {}); + await settled(); + + const [span] = harness.spans; + expect(span.ended).toBe(true); + expect(span.attributes).toEqual({}); + }); +}); + +describe("cloudflare telemetry span naming & attributes", () => { + it("names the span from the start payload and re-describes attributes at end", async () => { + // The srvx payload has no matched route or response at start โ€” the name is + // fixed to the concrete path. By `asyncEnd` the producers have populated + // `context.matchedRoute` and `result`, which land in the attributes. + const request: Record = { + method: "GET", + url: "https://example.com/users/123", + context: {} as Record, + }; + await traced("srvx.request", { request }, async () => { + (request.context as Record).matchedRoute = { route: "/users/:id" }; + return { status: 200 }; + }); + await settled(); + + const [span] = harness.spans; + expect(span.name).toBe("GET /users/123"); + expect(span.attributes["http.request.method"]).toBe("GET"); + expect(span.attributes["url.path"]).toBe("/users/123"); + expect(span.attributes["http.route"]).toBe("/users/:id"); + expect(span.attributes["http.response.status_code"]).toBe(200); + }); + + it("names h3.request spans by the matched route template", async () => { + const event = { + req: { method: "GET" }, + url: { pathname: "/users/123" }, + context: { matchedRoute: { route: "/users/:id" } }, + }; + await traced("h3.request", { type: "route", event }); + await settled(); + + const [span] = harness.spans; + expect(span.name).toBe("GET /users/:id"); + expect(span.attributes["h3.handler_type"]).toBe("route"); + expect(span.attributes["http.route"]).toBe("/users/:id"); + expect(span.attributes["url.path"]).toBe("/users/123"); + }); + + it("converts every OTLP attribute value variant", async () => { + await traced("unstorage.getItem", { + driver: { name: "redis" }, + base: "cache", + keys: ["a", "b"], + }); + await settled(); + + const [span] = harness.spans; + expect(span.attributes["db.system"]).toBe("redis"); // stringValue + expect(span.attributes["unstorage.keys_count"]).toBe(2); // intValue + }); +}); + +describe("cloudflare telemetry subscription", () => { + it("does not trace unknown/undeclared channels", async () => { + await expect(traced("ioredis:command", { command: { name: "GET" } })).resolves.toBe("ok"); + await settled(); + expect(harness.spans).toHaveLength(0); + }); + + it("drops the span on malformed payloads without breaking the operation", async () => { + // Missing `event` โ€” the h3 describer throws at start; no span is opened. + await expect(traced("h3.request", { type: "route" })).resolves.toBe("ok"); + await settled(); + expect(harness.spans).toHaveLength(0); + }); + + it("does not patch the global tracingChannel", () => { + expect(diagnostics.tracingChannel).toBe(originalTracingChannel); + }); + + it("traces a declared channel regardless of when the producer creates it", async () => { + diagnostics.tracingChannel("unstorage.removeItem"); + await traced("unstorage.removeItem", {}); + await settled(); + expect(harness.spans).toHaveLength(1); + expect(harness.spans[0].name).toBe("removeItem"); + }); +}); From 06d5d05de7088d3a07f4404b227cb5dd6b651d04 Mon Sep 17 00:00:00 2001 From: Rihan Arfan Date: Fri, 10 Jul 2026 13:16:35 +0100 Subject: [PATCH 2/6] fix(telemetry): report operations that throw synchronously --- src/runtime/internal/telemetry/subscribe.ts | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/src/runtime/internal/telemetry/subscribe.ts b/src/runtime/internal/telemetry/subscribe.ts index 9fc10abe3e..d18a1dbff9 100644 --- a/src/runtime/internal/telemetry/subscribe.ts +++ b/src/runtime/internal/telemetry/subscribe.ts @@ -12,7 +12,9 @@ export type SpanSink = (info: SpanInfo, startTimeUnixNano: string, error: unknow /** * Subscribes to the tracing channels declared in `TRACED_CHANNELS` (produced by - * h3, srvx, unstorage, โ€ฆ) and invokes `onSpan` for each completed operation. + * h3, srvx, unstorage, โ€ฆ) and invokes `onSpan` once per traced operation: + * normally at `asyncEnd`, or at `end` when the traced function threw + * synchronously (`tracePromise` never publishes `asyncEnd` in that case). * * A `tracingChannel()` publishes to plain named channels * (`tracing::start`, `tracing::asyncEnd`, โ€ฆ). Subscribing to those @@ -27,7 +29,7 @@ export function subscribeTracedChannels(onSpan: SpanSink): void { const diagnostics = globalThis.process?.getBuiltinModule?.("node:diagnostics_channel"); if (!diagnostics?.subscribe) return; - // Carry the start time from `start` to `asyncEnd` without mutating the producer's context object. + // Carry the start time from `start` to completion without mutating the producer's context object. const starts = new WeakMap(); for (const name of Object.keys(TRACED_CHANNELS)) { @@ -37,7 +39,7 @@ export function subscribeTracedChannels(onSpan: SpanSink): void { starts.set(message as object, Span.nowUnixNano()); }); - diagnostics.subscribe(`tracing:${name}:asyncEnd`, (message) => { + const complete = (message: unknown) => { try { const start = starts.get(message as object); if (start === undefined) return; @@ -52,6 +54,18 @@ export function subscribeTracedChannels(onSpan: SpanSink): void { } catch { // Malformed payload, or a sink failure โ€” never break the traced operation. } + }; + + diagnostics.subscribe(`tracing:${name}:asyncEnd`, complete); + + // `tracePromise` never publishes `asyncEnd` when the traced function + // throws synchronously โ€” only `end`, with `error` already set. In the + // normal async path `end` fires before the promise settles, while `error` + // is still unset, so the guard makes this a no-op there. + diagnostics.subscribe(`tracing:${name}:end`, (message) => { + if ((message as { error?: unknown }).error !== undefined) { + complete(message); + } }); } } From 7e90370a9befddad8ab1a88e28181c1aac14befb Mon Sep 17 00:00:00 2001 From: Rihan Arfan Date: Fri, 10 Jul 2026 13:17:04 +0100 Subject: [PATCH 3/6] feat(cloudflare): open observability spans via tracing channel onStart hook --- .../cloudflare/runtime/telemetry/plugin.ts | 76 +++++-------------- .../vercel/runtime/telemetry/plugin.ts | 3 + .../internal/telemetry/logger-plugin.ts | 10 ++- src/runtime/internal/telemetry/subscribe.ts | 73 +++++++++++++----- 4 files changed, 84 insertions(+), 78 deletions(-) diff --git a/src/presets/cloudflare/runtime/telemetry/plugin.ts b/src/presets/cloudflare/runtime/telemetry/plugin.ts index 9dff993495..4a1e9b2e52 100644 --- a/src/presets/cloudflare/runtime/telemetry/plugin.ts +++ b/src/presets/cloudflare/runtime/telemetry/plugin.ts @@ -3,7 +3,7 @@ import { definePlugin } from "nitro"; import * as cloudflare from "cloudflare:workers"; import type { Span, Tracing } from "@cloudflare/workers-types"; import type { IAnyValue } from "#nitro/runtime/telemetry/types"; -import { TRACED_CHANNELS } from "#nitro/runtime/telemetry/channels"; +import { subscribeTracedChannels } from "#nitro/runtime/telemetry/subscribe"; // https://developers.cloudflare.com/workers/observability/traces/custom-spans/ const tracing = (cloudflare as { tracing?: Tracing }).tracing; @@ -18,74 +18,40 @@ interface PendingSpan { * alongside Cloudflare's automatic instrumentation (fetch, KV, D1, โ€ฆ) */ export default definePlugin(() => { - const diagnostics = globalThis.process?.getBuiltinModule?.("node:diagnostics_channel"); - if (!diagnostics?.subscribe || typeof tracing?.enterSpan !== "function") return; + if (typeof tracing?.enterSpan !== "function") return; - // Open span + closer per in-flight operation, keyed by the context object - // `tracingChannel` publishes to every phase of the same operation. - const pending = new WeakMap(); - - for (const name of Object.keys(TRACED_CHANNELS)) { - const describe = TRACED_CHANNELS[name]; - - diagnostics.subscribe(`tracing:${name}:start`, (message) => { - try { - // The span name is fixed at creation (Cloudflare has no rename API), - // so it is derived from the start payload; end-time data (status code, - // matched route) lands in the attributes set at `asyncEnd`. - const info = describe(name, message); - - let close!: () => void; - const done = new Promise((resolve) => { - close = resolve; - }); - tracing.enterSpan(info.name, (span) => { - pending.set(message as object, { span, close }); - return done; - }); - } catch { - // Malformed payload, or tracing rejected the span (e.g. outside a - // request) โ€” skip it, never break the traced operation. - } - }); - - const finalize = (message: unknown) => { - const entry = pending.get(message as object); + subscribeTracedChannels( + (info, _startTimeUnixNano, error, entry) => { if (!entry) return; - pending.delete(message as object); try { // Skip attribute work for unsampled requests (`head_sampling_rate`). - if (entry.span.isTraced) { - // Re-describe on the completed payload: by now producers have set - // end-time fields (`result.status`, `context.matchedRoute`, โ€ฆ). - const info = describe(name, message); + if (info && entry.span.isTraced) { for (const { key, value } of info.attributes) { entry.span.setAttribute(key, attributeValue(value)); } - const error = (message as { error?: unknown }).error; if (error !== undefined) { recordException(entry.span, error); } } - } catch { - // Never break the traced operation. } finally { entry.close(); } - }; - - diagnostics.subscribe(`tracing:${name}:asyncEnd`, finalize); - - // `tracePromise` never publishes `asyncEnd` when the traced function - // throws synchronously โ€” only `end`, with `error` already set. In the - // normal async path `end` fires before the promise settles, while `error` - // is still unset, so the guard makes this a no-op there. - diagnostics.subscribe(`tracing:${name}:end`, (message) => { - if ((message as { error?: unknown }).error !== undefined) { - finalize(message); - } - }); - } + }, + { + onStart(info) { + let close!: () => void; + const done = new Promise((resolve) => { + close = resolve; + }); + let entry: PendingSpan | undefined; + tracing.enterSpan(info.name, (span) => { + entry = { span, close }; + return done; + }); + return entry; + }, + } + ); }); /** OTLP `IAnyValue` (from the shared describers) โ†’ Cloudflare attribute value. */ diff --git a/src/presets/vercel/runtime/telemetry/plugin.ts b/src/presets/vercel/runtime/telemetry/plugin.ts index 7f494ac463..a69bc99e8d 100644 --- a/src/presets/vercel/runtime/telemetry/plugin.ts +++ b/src/presets/vercel/runtime/telemetry/plugin.ts @@ -47,6 +47,9 @@ const pendingSpans = new Map(); */ export default definePlugin(() => { subscribeTracedChannels((info, startTimeUnixNano, error) => { + // Describer failed on the completed payload โ€” nothing to report. + if (!info) return; + const context = (globalThis as Record)[ REQUEST_CONTEXT_SYMBOL ]?.get?.(); diff --git a/src/runtime/internal/telemetry/logger-plugin.ts b/src/runtime/internal/telemetry/logger-plugin.ts index b44580ea11..8f21989bde 100644 --- a/src/runtime/internal/telemetry/logger-plugin.ts +++ b/src/runtime/internal/telemetry/logger-plugin.ts @@ -61,9 +61,10 @@ export default definePlugin((nitroApp: NitroApp) => { subscribeTracedChannels((info, start, error) => { const trace = als.getStore(); - // No active request (e.g. storage during plugin setup, or the platform's - // outer request span that closes after the response) โ€” nothing to attach to. - if (!trace) return; + // No span info (describer failed on the payload), or no active request + // (e.g. storage during plugin setup, or the platform's outer request span + // that closes after the response) โ€” nothing to attach to. + if (!info || !trace) return; trace.spans.push({ name: info.name, attributes: info.attributes, @@ -321,7 +322,8 @@ function ms(nanos: bigint): string { return `${(Number(nanos) / 1e6).toFixed(2)}ms`; } -function logFlat(info: SpanInfo, startTimeUnixNano: string, error: unknown): void { +function logFlat(info: SpanInfo | undefined, startTimeUnixNano: string, error: unknown): void { + if (!info) return; const durationMs = Number(BigInt(Span.nowUnixNano()) - BigInt(startTimeUnixNano)) / 1e6; const attrs = info.attributes.length ? ` ${formatAttributes(info.attributes)}` : ""; const line = `[trace] ${info.name} (${durationMs.toFixed(2)}ms)${attrs}`; diff --git a/src/runtime/internal/telemetry/subscribe.ts b/src/runtime/internal/telemetry/subscribe.ts index d18a1dbff9..7be8673cff 100644 --- a/src/runtime/internal/telemetry/subscribe.ts +++ b/src/runtime/internal/telemetry/subscribe.ts @@ -3,12 +3,28 @@ import { TRACED_CHANNELS } from "./channels.ts"; import { Span } from "./span.ts"; /** - * Called once per completed traced operation with the derived span info, the - * operation start time (unix nanoseconds, as an OTLP `*UnixNano` string) and the - * operation's error (`undefined` when it succeeded). A sink turns this into - * whatever its platform consumes โ€” an OTLP export, a log line, โ€ฆ + * Called once per completed traced operation with the span info derived from + * the completed payload (`undefined` when the describer failed on it), the + * operation start time (unix nanoseconds, as an OTLP `*UnixNano` string), the + * operation's error (`undefined` when it succeeded) and the state returned by + * `onStart` (`undefined` without one). A sink turns this into whatever its + * platform consumes โ€” an OTLP export, a log line, a platform span, โ€ฆ */ -export type SpanSink = (info: SpanInfo, startTimeUnixNano: string, error: unknown) => void; +export type SpanSink = ( + info: SpanInfo | undefined, + startTimeUnixNano: string, + error: unknown, + state: S | undefined +) => void; + +export interface SubscribeTracedChannelsOptions { + /** + * Called synchronously when a traced operation starts, inside its execution + * context โ€” where platform span APIs like Cloudflare's `enterSpan` must be + * called. Returned state is handed back to `onSpan` at completion. + */ + onStart?: (info: SpanInfo) => S | undefined; +} /** * Subscribes to the tracing channels declared in `TRACED_CHANNELS` (produced by @@ -25,34 +41,53 @@ export type SpanSink = (info: SpanInfo, startTimeUnixNano: string, error: unknow * * A no-op when `node:diagnostics_channel` is unavailable (non-Node runtimes). */ -export function subscribeTracedChannels(onSpan: SpanSink): void { +export function subscribeTracedChannels( + onSpan: SpanSink, + options?: SubscribeTracedChannelsOptions +): void { const diagnostics = globalThis.process?.getBuiltinModule?.("node:diagnostics_channel"); if (!diagnostics?.subscribe) return; - // Carry the start time from `start` to completion without mutating the producer's context object. - const starts = new WeakMap(); + const onStart = options?.onStart; + + // Carry the start time (and any sink state) from `start` to completion + // without mutating the producer's context object. + const pending = new WeakMap(); for (const name of Object.keys(TRACED_CHANNELS)) { const describe = TRACED_CHANNELS[name]; diagnostics.subscribe(`tracing:${name}:start`, (message) => { - starts.set(message as object, Span.nowUnixNano()); + const entry = { start: Span.nowUnixNano(), state: undefined as S | undefined }; + if (onStart) { + try { + entry.state = onStart(describe(name, message)); + } catch { + // Malformed payload, or a sink failure (e.g. the platform refused to + // open a span) โ€” no state; the completion callback still fires. + } + } + pending.set(message as object, entry); }); const complete = (message: unknown) => { + const entry = pending.get(message as object); + if (entry === undefined) return; + pending.delete(message as object); + + // Derive span name, kind and semantic attributes from the completed + // operation. A describer only throws on a payload shape it doesn't + // recognise (a producer that changed shape); still deliver the + // completion โ€” without info โ€” so stateful sinks can release their span. + let info: SpanInfo | undefined; try { - const start = starts.get(message as object); - if (start === undefined) return; - starts.delete(message as object); + info = describe(name, message); + } catch {} - // Derive span name, kind and semantic attributes from the operation. A - // describer only throws on a payload shape it doesn't recognise (a - // producer that changed shape); drop that span via the catch below - // rather than emit a contentless one. - const info = describe(name, message); - onSpan(info, start, (message as { error?: unknown }).error); + try { + onSpan(info, entry.start, (message as { error?: unknown }).error, entry.state); } catch { - // Malformed payload, or a sink failure โ€” never break the traced operation. + // A sink failure must never break the traced operation. } }; From cc93487c67b0ddd3040bfb53721a0a638075a887 Mon Sep 17 00:00:00 2001 From: Rihan Arfan Date: Fri, 10 Jul 2026 17:32:01 +0100 Subject: [PATCH 4/6] test(telemetry): cover subscribeTracedChannels --- test/unit/telemetry-subscribe.test.ts | 173 ++++++++++++++++++++++++++ 1 file changed, 173 insertions(+) create mode 100644 test/unit/telemetry-subscribe.test.ts diff --git a/test/unit/telemetry-subscribe.test.ts b/test/unit/telemetry-subscribe.test.ts new file mode 100644 index 0000000000..391de4d5f8 --- /dev/null +++ b/test/unit/telemetry-subscribe.test.ts @@ -0,0 +1,173 @@ +import { beforeAll, beforeEach, describe, expect, it } from "vitest"; +import type { TracingChannel } from "node:diagnostics_channel"; +import type { SpanInfo } from "../../src/runtime/internal/telemetry/types.ts"; +import { subscribeTracedChannels } from "../../src/runtime/internal/telemetry/subscribe.ts"; + +const diagnostics = process.getBuiltinModule("node:diagnostics_channel"); + +interface Completion { + info: SpanInfo | undefined; + start: string; + error: unknown; + state: unknown; +} + +// --------------------------------------------------------------------------- +// Setup: `diagnostics_channel.subscribe` has no per-test teardown, so a single +// module-level subscription records every completion and delegates the +// per-test behavior (`onStart` return value / sink failures) to swappable +// implementations. +// --------------------------------------------------------------------------- + +const completions: Completion[] = []; +const startInfos: SpanInfo[] = []; +let onStartImpl: ((info: SpanInfo) => unknown) | undefined; +let sinkImpl: ((completion: Completion) => void) | undefined; + +beforeAll(() => { + subscribeTracedChannels( + (info, start, error, state) => { + const completion: Completion = { info, start, error, state }; + completions.push(completion); + sinkImpl?.(completion); + }, + { + onStart(info) { + startInfos.push(info); + return onStartImpl?.(info); + }, + } + ); +}); + +beforeEach(() => { + completions.length = 0; + startInfos.length = 0; + onStartImpl = undefined; + sinkImpl = undefined; +}); + +/** Run one traced operation the way producers (h3/srvx/unstorage) do. */ +function traced( + channelName: string, + data: Record, + fn: () => Promise = async () => "ok" +): Promise { + const channel = diagnostics.tracingChannel(channelName) as TracingChannel< + Record + >; + return channel.tracePromise(fn, data); +} + +/** Let any stray `asyncEnd` continuation land before asserting counts. */ +function settled(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe("subscribeTracedChannels onStart state", () => { + it("hands the state returned by onStart back to the sink at completion", async () => { + const token = { platformSpan: true }; + onStartImpl = () => token; + + await traced("unstorage.setItem", { base: "cache" }); + + expect(startInfos).toHaveLength(1); + expect(startInfos[0].name).toBe("setItem cache"); + expect(completions).toHaveLength(1); + expect(completions[0].state).toBe(token); + expect(completions[0].info?.name).toBe("setItem cache"); + expect(completions[0].error).toBeUndefined(); + }); + + it("still delivers the completion when onStart throws", async () => { + onStartImpl = () => { + throw new Error("platform refused the span"); + }; + + await expect(traced("unstorage.setItem", {})).resolves.toBe("ok"); + + expect(completions).toHaveLength(1); + expect(completions[0].state).toBeUndefined(); + expect(completions[0].info?.name).toBe("setItem"); + }); + + it("skips onStart on a malformed start payload but still completes", async () => { + // Missing `event` โ€” the h3 describer throws at start, so `onStart` never + // runs; by completion the producer has populated the payload. + const data: Record = { type: "route" }; + await traced("h3.request", data, async () => { + data.event = { req: { method: "GET" }, url: { pathname: "/x" } }; + return "ok"; + }); + + expect(startInfos).toHaveLength(0); + expect(completions).toHaveLength(1); + expect(completions[0].info?.name).toBe("GET /x"); + expect(completions[0].state).toBeUndefined(); + }); + + it("delivers the completion without info when the describer fails on the completed payload", async () => { + const token = { platformSpan: true }; + onStartImpl = () => token; + + const data: Record = { + type: "route", + event: { req: { method: "GET" }, url: { pathname: "/x" } }, + }; + await traced("h3.request", data, async () => { + // The end-time payload no longer matches the describer's shape. + delete data.event; + return "ok"; + }); + + expect(startInfos).toHaveLength(1); + expect(completions).toHaveLength(1); + expect(completions[0].info).toBeUndefined(); + // Stateful sinks still get their state back to release the platform span. + expect(completions[0].state).toBe(token); + }); +}); + +describe("subscribeTracedChannels completion semantics", () => { + it("reports an operation that throws synchronously exactly once", async () => { + // A sync throw makes `tracePromise` publish `end` (with `error` set) and + // rethrow without ever publishing `asyncEnd`. + const error = new Error("sync failure"); + expect(() => + traced("unstorage.setItem", {}, () => { + throw error; + }) + ).toThrow("sync failure"); + await settled(); + + expect(completions).toHaveLength(1); + expect(completions[0].error).toBe(error); + expect(completions[0].info?.name).toBe("setItem"); + }); + + it("reports a rejected operation exactly once, at asyncEnd", async () => { + const error = new Error("async failure"); + await expect( + traced("unstorage.setItem", {}, async () => { + throw error; + }) + ).rejects.toThrow("async failure"); + await settled(); + + expect(completions).toHaveLength(1); + expect(completions[0].error).toBe(error); + }); + + it("never breaks the traced operation when the sink throws", async () => { + sinkImpl = () => { + throw new Error("sink failure"); + }; + + await expect(traced("unstorage.getItem", {})).resolves.toBe("ok"); + expect(completions).toHaveLength(1); + }); +}); From e259c68ef03a2c5a9592a7f8f61ae149bc33e2ab Mon Sep 17 00:00:00 2001 From: Rihan Arfan Date: Fri, 10 Jul 2026 17:32:11 +0100 Subject: [PATCH 5/6] fix(cloudflare): record exceptions even when the describer fails at completion --- test/unit/cloudflare-telemetry.test.ts | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/test/unit/cloudflare-telemetry.test.ts b/test/unit/cloudflare-telemetry.test.ts index 03c1b5833b..dc1683697d 100644 --- a/test/unit/cloudflare-telemetry.test.ts +++ b/test/unit/cloudflare-telemetry.test.ts @@ -184,6 +184,29 @@ describe("cloudflare telemetry span lifecycle", () => { expect(span.attributes["exception.message"]).toBe("sync failure"); }); + it("records the exception even when the describer fails on the completed payload", async () => { + const error = new Error("late failure"); + const data: Record = { + type: "route", + event: { req: { method: "GET" }, url: { pathname: "/x" } }, + }; + await expect( + traced("h3.request", data, async () => { + // The end-time payload no longer matches the describer's shape. + delete data.event; + throw error; + }) + ).rejects.toThrow("late failure"); + await settled(); + + const [span] = harness.spans; + expect(span.ended).toBe(true); + // Named at start, no end-time attributes โ€” but the error is recorded. + expect(span.name).toBe("GET /x"); + expect(span.attributes["http.request.method"]).toBeUndefined(); + expect(span.attributes["exception.message"]).toBe("late failure"); + }); + it("records non-Error rejections as exception.message only", async () => { await expect( traced("unstorage.setItem", {}, async () => { From 0668713b77913cd71e4fabe7f77d637492a6379e Mon Sep 17 00:00:00 2001 From: Rihan Arfan Date: Fri, 10 Jul 2026 17:32:57 +0100 Subject: [PATCH 6/6] fix(cloudflare): record exceptions even when the describer fails at completion --- src/presets/cloudflare/runtime/telemetry/plugin.ts | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/presets/cloudflare/runtime/telemetry/plugin.ts b/src/presets/cloudflare/runtime/telemetry/plugin.ts index 4a1e9b2e52..12e6e2fa6c 100644 --- a/src/presets/cloudflare/runtime/telemetry/plugin.ts +++ b/src/presets/cloudflare/runtime/telemetry/plugin.ts @@ -25,9 +25,13 @@ export default definePlugin(() => { if (!entry) return; try { // Skip attribute work for unsampled requests (`head_sampling_rate`). - if (info && entry.span.isTraced) { - for (const { key, value } of info.attributes) { - entry.span.setAttribute(key, attributeValue(value)); + if (entry.span.isTraced) { + // `info` is undefined when the describer failed on the completed + // payload โ€” the error is still recorded on the (already named) span. + if (info) { + for (const { key, value } of info.attributes) { + entry.span.setAttribute(key, attributeValue(value)); + } } if (error !== undefined) { recordException(entry.span, error);