@@ -3,16 +3,35 @@ import { TRACED_CHANNELS } from "./channels.ts";
33import { Span } from "./span.ts" ;
44
55/**
6- * Called once per completed traced operation with the derived span info, the
7- * operation start time (unix nanoseconds, as an OTLP `*UnixNano` string) and the
8- * operation's error (`undefined` when it succeeded). A sink turns this into
9- * whatever its platform consumes — an OTLP export, a log line, …
6+ * Called once per completed traced operation with the span info (derived from
7+ * the completed payload, falling back to the start-time payload for `onStart`
8+ * subscriptions), the operation start time (unix nanoseconds, as an OTLP
9+ * `*UnixNano` string), the operation's error (`undefined` when it succeeded)
10+ * and the state returned by `onStart` (`undefined` without one). A sink turns
11+ * this into whatever its platform consumes — an OTLP export, a log line, a
12+ * platform span, …
1013 */
11- export type SpanSink = ( info : SpanInfo , startTimeUnixNano : string , error : unknown ) => void ;
14+ export type SpanSink < S = unknown > = (
15+ info : SpanInfo ,
16+ startTimeUnixNano : string ,
17+ error : unknown ,
18+ state : S | undefined
19+ ) => void ;
20+
21+ export interface SubscribeTracedChannelsOptions < S > {
22+ /**
23+ * Called synchronously when a traced operation starts, inside its execution
24+ * context — where platform span APIs like Cloudflare's `enterSpan` must be
25+ * called. Returned state is handed back to `onSpan` at completion.
26+ */
27+ onStart ?: ( info : SpanInfo ) => S | undefined ;
28+ }
1229
1330/**
1431 * Subscribes to the tracing channels declared in `TRACED_CHANNELS` (produced by
15- * h3, srvx, unstorage, …) and invokes `onSpan` for each completed operation.
32+ * h3, srvx, unstorage, …) and invokes `onSpan` once per traced operation:
33+ * normally at `asyncEnd`, or at `end` when the traced function threw
34+ * synchronously (`tracePromise` never publishes `asyncEnd` in that case).
1635 *
1736 * A `tracingChannel(<name>)` publishes to plain named channels
1837 * (`tracing:<name>:start`, `tracing:<name>:asyncEnd`, …). Subscribing to those
@@ -23,34 +42,74 @@ export type SpanSink = (info: SpanInfo, startTimeUnixNano: string, error: unknow
2342 *
2443 * A no-op when `node:diagnostics_channel` is unavailable (non-Node runtimes).
2544 */
26- export function subscribeTracedChannels ( onSpan : SpanSink ) : void {
45+ export function subscribeTracedChannels < S = unknown > (
46+ onSpan : SpanSink < S > ,
47+ options ?: SubscribeTracedChannelsOptions < S >
48+ ) : void {
2749 const diagnostics = globalThis . process ?. getBuiltinModule ?.( "node:diagnostics_channel" ) ;
2850 if ( ! diagnostics ?. subscribe ) return ;
2951
30- // Carry the start time from `start` to `asyncEnd` without mutating the producer's context object.
31- const starts = new WeakMap < object , string > ( ) ;
52+ const onStart = options ?. onStart ;
53+
54+ // Carry the start time (and any start-time info / sink state) from `start`
55+ // to completion without mutating the producer's context object.
56+ interface Pending {
57+ start : string ;
58+ info : SpanInfo | undefined ;
59+ state : S | undefined ;
60+ }
61+ const pending = new WeakMap < object , Pending > ( ) ;
3262
3363 for ( const name of Object . keys ( TRACED_CHANNELS ) ) {
3464 const describe = TRACED_CHANNELS [ name ] ;
3565
3666 diagnostics . subscribe ( `tracing:${ name } , ( message ) => {
37- starts . set ( message as object , Span . nowUnixNano ( ) ) ;
67+ const entry : Pending = { start : Span . nowUnixNano ( ) , info : undefined , state : undefined } ;
68+ if ( onStart ) {
69+ try {
70+ entry . info = describe ( name , message ) ;
71+ entry . state = onStart ( entry . info ) ;
72+ } catch {
73+ // Malformed payload, or a sink failure (e.g. the platform refused to
74+ // open a span) — no state; the completion callback still fires.
75+ }
76+ }
77+ pending . set ( message as object , entry ) ;
3878 } ) ;
3979
40- diagnostics . subscribe ( `tracing:${ name } , ( message ) => {
80+ const complete = ( message : unknown ) => {
81+ const entry = pending . get ( message as object ) ;
82+ if ( entry === undefined ) return ;
83+ pending . delete ( message as object ) ;
84+
85+ // Derive span name, kind and semantic attributes from the completed
86+ // operation. A describer only throws on a payload shape it doesn't
87+ // recognise (a producer that changed shape); fall back to the start-time
88+ // info (held for `onStart` subscriptions) so stateful sinks still get
89+ // the completion and can release their span.
90+ let info : SpanInfo | undefined ;
4191 try {
42- const start = starts . get ( message as object ) ;
43- if ( start === undefined ) return ;
44- starts . delete ( message as object ) ;
45-
46- // Derive span name, kind and semantic attributes from the operation. A
47- // describer only throws on a payload shape it doesn't recognise (a
48- // producer that changed shape); drop that span via the catch below
49- // rather than emit a contentless one.
50- const info = describe ( name , message ) ;
51- onSpan ( info , start , ( message as { error ?: unknown } ) . error ) ;
92+ info = describe ( name , message ) ;
93+ } catch { }
94+ info ??= entry . info ;
95+ if ( info === undefined ) return ;
96+
97+ try {
98+ onSpan ( info , entry . start , ( message as { error ?: unknown } ) . error , entry . state ) ;
5299 } catch {
53- // Malformed payload, or a sink failure — never break the traced operation.
100+ // A sink failure must never break the traced operation.
101+ }
102+ } ;
103+
104+ diagnostics . subscribe ( `tracing:${ name } , complete ) ;
105+
106+ // `tracePromise` never publishes `asyncEnd` when the traced function
107+ // throws synchronously — only `end`, with `error` already set. In the
108+ // normal async path `end` fires before the promise settles, while `error`
109+ // is still unset, so the guard makes this a no-op there.
110+ diagnostics . subscribe ( `tracing:${ name } , ( message ) => {
111+ if ( ( message as { error ?: unknown } ) . error !== undefined ) {
112+ complete ( message ) ;
54113 }
55114 } ) ;
56115 }
0 commit comments