From 4b95fd730b53264c90264497305d39473f7f7624 Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 10 Jun 2026 12:10:06 +0100 Subject: [PATCH 1/6] Add tracing module --- core/package.json | 6 + core/src/tracing/index.ts | 99 +++++++++++ core/src/tracing/opentelemetry.ts | 236 ++++++++++++++++++++++++++ core/src/tracing/proxy.ts | 131 ++++++++++++++ core/src/tracing/simple.ts | 272 ++++++++++++++++++++++++++++++ pnpm-lock.yaml | 89 ++++++++++ 6 files changed, 833 insertions(+) create mode 100644 core/src/tracing/index.ts create mode 100644 core/src/tracing/opentelemetry.ts create mode 100644 core/src/tracing/proxy.ts create mode 100644 core/src/tracing/simple.ts diff --git a/core/package.json b/core/package.json index 7bcd397..f02568f 100644 --- a/core/package.json +++ b/core/package.json @@ -23,6 +23,12 @@ "prepack": "pnpm build" }, "dependencies": { + "@opentelemetry/api": "^1.9.1", + "@opentelemetry/context-async-hooks": "^2.7.1", + "@opentelemetry/resources": "^2.7.1", + "@opentelemetry/sdk-trace-base": "^2.7.1", + "@opentelemetry/sdk-trace-node": "^2.7.1", + "@opentelemetry/semantic-conventions": "^1.41.1", "@optique/core": "^0.6.2", "@optique/run": "^0.6.2", "fluture": "^14.0.0", diff --git a/core/src/tracing/index.ts b/core/src/tracing/index.ts new file mode 100644 index 0000000..fc882df --- /dev/null +++ b/core/src/tracing/index.ts @@ -0,0 +1,99 @@ +// Tracing facade. +// +// The application imports `trace`, `traceP`, `traceF` and `traceRequests` from +// here. The first three delegate to the currently-selected `Tracer`, so the +// whole program's tracing destination is switched at runtime with `setTracer`. +// +// There is NO tracer by default: Use `setTracer` to set one. +// +export { type Tracer, type Attributes, type Name, setTracer, trace, traceP, traceF, event }; + +import type { Future } from "../future"; + +// A flat bag of primitive values attached to a trace or event. +type Attributes = Record; + +// The first argument to a trace function: either a plain name, or an attributes +// object that carries the name under its `name` field. The `name` becomes the +// trace's name; the remaining keys become its attributes. +type Name = string | (Attributes & { name: string }); + +// The common shape of a tracer. The facade normalises the public `Name` argument +// into a name + attributes before delegating, so implementations receive both +// explicitly (no polyvariadic signatures, no trailing attributes). +interface Tracer { + /** + * Trace a synchronous call. + */ + trace(name: string, attributes: Attributes, f: () => A): A; + + /** + * Trace an asynchronous call. + */ + traceP(name: string, attributes: Attributes, f: () => Promise): Promise; + + /** + * Trace a Future + */ + traceF(name: string, attributes: Attributes, f: Future): Future; + + /** + * Record a point-in-time event on the currently-active trace. + */ + event(name: string, attributes?: Attributes): void; +} + +// The active tracer. None by default — traces are no-ops until one is set. +let current: Tracer | undefined = undefined; + +// Choose where traces go. Affects every subsequent trace call across the program. +function setTracer(tracer: Tracer): void { + current = tracer; +} + +// The delegating API. These read `current` at call time, so a `setTracer` switch +// takes effect immediately for all callers. With no tracer set, they run the +// work untraced. The name argument may be a plain string or an attributes object +// carrying the name (see `Name`). +function trace(name: Name, f: () => A): A { + if (!current) { + return f(); + } + const t = named(name); + return current.trace(t.name, t.attributes, f); +} + +function traceP(name: Name, f: () => Promise): Promise { + if (!current) { + return f(); + } + const t = named(name); + return current.traceP(t.name, t.attributes, f); +} + +function traceF(name: Name, f: Future): Future { + if (!current) { + return f; + } + const t = named(name); + return current.traceF(t.name, t.attributes, f); +} + +// Record a named, timestamped marker on whatever trace is currently active — for +// annotating "something happened here" (a cache miss, a retry, a validation +// failure) at a point in time, without opening a child trace. A no-op when no +// trace is active (or no tracer is set). +function event(name: string, attributes?: Attributes): void { + current?.event(name, attributes); +} + +// Split a `Name` into the trace name and its attributes. +function named(name: Name): { name: string; attributes: Attributes } { + if (typeof name === "string") { + return { name, attributes: {} }; + } + const { name: n, ...attributes } = name; + return { name: n, attributes }; +} + + diff --git a/core/src/tracing/opentelemetry.ts b/core/src/tracing/opentelemetry.ts new file mode 100644 index 0000000..661aea8 --- /dev/null +++ b/core/src/tracing/opentelemetry.ts @@ -0,0 +1,236 @@ +// OpenTelemetry tracing: a `Tracer` (see ./index.ts) exposing the same API as +// `./simple.ts` — `trace`, `traceP`, `traceF`. +// +// Nesting is automatic and needs no token threaded through the code — it rides on +// OpenTelemetry's AsyncLocalStorage context manager, which propagates the active +// span across `await`s and Fluture's `.chain`/`.map` continuations. A span +// created while another is active becomes its child. +// +// Construct it with a service name and a span processor: +// +// new OpenTelemetryTracer({ serviceName: "backend", spanProcessor: new OtlpJsonStdoutProcessor() }) +// +// `OtlpJsonStdoutProcessor` (also exported here) prints each completed trace to +// stdout as an OTLP/JSON document (`{ resourceSpans: [...] }`) — paste it into an +// OTLP trace viewer such as https://tracekit.dev/tools/trace-visualizer. +export { OpenTelemetryTracer, OtlpJsonStdoutProcessor }; + +import { type Tracer, type Attributes } from "./index"; +import { Future } from "../future"; +import { + context, + trace as otel, + type Context, + type Span, + type Tracer as OtelTracer, + type HrTime, + type AttributeValue, +} from "@opentelemetry/api"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; +import type { SpanProcessor, ReadableSpan } from "@opentelemetry/sdk-trace-base"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { resourceFromAttributes } from "@opentelemetry/resources"; +import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions"; + +// --- OTLP/JSON stdout span processor --------------------------------------- +// +// Prints completed traces to stdout as OTLP/JSON (`{ resourceSpans: [...] }`), +// one self-contained document per trace, flushed when the trace's ROOT span ends +// (which is after all of its children). Hand-rolled to avoid depending on +// @opentelemetry/otlp-transformer. trace/span ids are emitted as hex and int +// attributes as strings, per the OTLP/JSON encoding the OTel SDKs and Collector +// use. + +class OtlpJsonStdoutProcessor implements SpanProcessor { + // Spans accumulated per in-flight trace, flushed when the trace's root ends. + private readonly byTrace = new Map(); + + onStart(): void {} + + onEnd(span: ReadableSpan): void { + const traceId = span.spanContext().traceId; + const spans = this.byTrace.get(traceId) ?? []; + spans.push(span); + this.byTrace.set(traceId, spans); + + if (!parentSpanId(span)) { + // Root span ended → the trace is complete. + this.byTrace.delete(traceId); + console.log(JSON.stringify(toOtlp(spans))); + } + } + + forceFlush(): Promise { + return Promise.resolve(); + } + + shutdown(): Promise { + return Promise.resolve(); + } +} + +// ReadableSpan exposes the parent either as `parentSpanContext` (newer SDKs) or +// `parentSpanId` (older ones). A missing/empty value means this is a root span. +function parentSpanId(span: ReadableSpan): string | undefined { + const s = span as unknown as { parentSpanContext?: { spanId?: string }; parentSpanId?: string }; + return s.parentSpanContext?.spanId ?? s.parentSpanId ?? undefined; +} + +function toOtlp(spans: ReadableSpan[]): unknown { + const first = spans[0]; + return { + resourceSpans: [ + { + resource: { attributes: toAttributes(first?.resource.attributes ?? {}) }, + scopeSpans: [ + { + scope: { + name: first?.instrumentationScope.name ?? "", + version: first?.instrumentationScope.version, + }, + spans: spans.map(toOtlpSpan), + }, + ], + }, + ], + }; +} + +function toOtlpSpan(span: ReadableSpan): unknown { + const ctx = span.spanContext(); + return { + traceId: ctx.traceId, + spanId: ctx.spanId, + parentSpanId: parentSpanId(span), + name: span.name, + // OTLP SpanKind is the API SpanKind + 1 (OTLP reserves 0 for UNSPECIFIED). + kind: span.kind + 1, + startTimeUnixNano: hrTimeToNanoString(span.startTime), + endTimeUnixNano: hrTimeToNanoString(span.endTime), + attributes: toAttributes(span.attributes), + events: span.events.map(e => ({ + timeUnixNano: hrTimeToNanoString(e.time), + name: e.name, + attributes: toAttributes(e.attributes ?? {}), + })), + status: { code: span.status.code }, + }; +} + +function toAttributes(attrs: Record): unknown[] { + return Object.entries(attrs) + .filter(([, v]) => v !== undefined) + .map(([key, value]) => ({ key, value: toAnyValue(value as AttributeValue) })); +} + +function toAnyValue(value: AttributeValue): unknown { + if (typeof value === "string") { + return { stringValue: value }; + } + if (typeof value === "boolean") { + return { boolValue: value }; + } + if (typeof value === "number") { + // int64 fields are encoded as strings in OTLP/JSON. + return Number.isInteger(value) ? { intValue: String(value) } : { doubleValue: value }; + } + if (Array.isArray(value)) { + return { arrayValue: { values: value.map(v => toAnyValue(v as AttributeValue)) } }; + } + return { stringValue: String(value) }; +} + +// HrTime is [seconds, nanos]. Combine via BigInt to avoid the precision loss of +// representing epoch-nanoseconds (~1.8e18) as a JS number. +function hrTimeToNanoString(time: HrTime): string { + return (BigInt(time[0]) * 1_000_000_000n + BigInt(time[1])).toString(); +} + +// --- The OpenTelemetry tracer ---------------------------------------------- + +interface OpenTelemetryTracerOptions { + // Logical name of the service, recorded as the resource `service.name`. + serviceName: string; + // Where finished spans go. Use `new OtlpJsonStdoutProcessor()` for the stdout + // OTLP/JSON output, or a BatchSpanProcessor + OTLP exporter for a backend. + spanProcessor: SpanProcessor; +} + +class OpenTelemetryTracer implements Tracer { + private readonly tracer: OtelTracer; + + constructor(options: OpenTelemetryTracerOptions) { + const provider = new NodeTracerProvider({ + resource: resourceFromAttributes({ [ATTR_SERVICE_NAME]: options.serviceName }), + spanProcessors: [options.spanProcessor], + }); + // The AsyncLocalStorage context manager is what lets `context.active()` + // return the right parent span inside `await`s and Fluture continuations. + provider.register({ contextManager: new AsyncLocalStorageContextManager().enable() }); + this.tracer = otel.getTracer(options.serviceName); + } + + // Begin a span as a child of the active one (or a new root when none is + // active) and return both the span and the context that makes it active. + private begin(name: string, attributes: Attributes): { span: Span; ctx: Context } { + const parentCtx = context.active(); + const span = this.tracer.startSpan(name, { attributes }, parentCtx); + return { span, ctx: otel.setSpan(parentCtx, span) }; + } + + // Trace a synchronous function. + trace(name: string, attributes: Attributes, f: () => A): A { + const { span, ctx } = this.begin(name, attributes); + try { + return context.with(ctx, f); + } finally { + span.end(); + } + } + + // Trace an asynchronous function. `context.with` keeps the span active across + // the awaits inside `f` (AsyncLocalStorage propagates it), so spans created + // within `f` nest under it. After the awaited promise settles we are back in + // the caller's context, so the span ends in the right place and a subsequent + // `traceP` becomes a sibling. + async traceP(name: string, attributes: Attributes, f: () => Promise): Promise { + const { span, ctx } = this.begin(name, attributes); + try { + return await context.with(ctx, f); + } finally { + span.end(); + } + } + + // Trace a Future. Because Futures are lazy, the span begins when the Future is + // forked (its real start) and ends when it settles. The span is made active + // for the Future's execution so nested spans nest under it; the parent context + // is restored when settling so that a `.chain` *after* `traceF(...)` is a + // sibling rather than a descendant. + traceF(name: string, attributes: Attributes, f: Future): Future { + return Future.create((reject, resolve) => { + const parentCtx = context.active(); + const span = this.tracer.startSpan(name, { attributes }, parentCtx); + const ctx = otel.setSpan(parentCtx, span); + return context.with(ctx, () => + f.fork( + err => { + span.end(); + context.with(parentCtx, () => reject(err)); + }, + val => { + span.end(); + context.with(parentCtx, () => resolve(val)); + }, + ), + ); + }); + } + + // Record a point-in-time event on the currently-active span (the span made + // active by an enclosing trace/traceP/traceF). No-op when none is active. + event(name: string, attributes?: Attributes): void { + otel.getActiveSpan()?.addEvent(name, attributes); + } +} + diff --git a/core/src/tracing/proxy.ts b/core/src/tracing/proxy.ts new file mode 100644 index 0000000..18ba7c1 --- /dev/null +++ b/core/src/tracing/proxy.ts @@ -0,0 +1,131 @@ +// A Tracer that wraps another Tracer and forwards only the traces whose name +// passes the configured filters. Non-matching traces run untraced — the work +// still executes, but no span is created. +// +// The match is decided once per call and applies to the whole span (delegate the +// call wholesale, or bypass it wholesale), so a span is never half-emitted. +// +// Optionally (config.includeCallSite) it tags each forwarded trace with the +// source file and line it was called from, as `code.filepath` / `code.lineno`. +// +// Usage: +// setTracer(new ProxyTracer({ +// tracer: new OpenTelemetryTracer({ ... }), +// traceFilter: name => /^mongo\./.test(name) || name.includes("idempotency"), +// includeCallSite: true, +// })); +export { ProxyTracer, type Config }; + +import { type Tracer, type Attributes } from "./index"; +import type { Future } from "../future"; +import { fileURLToPath } from "node:url"; +import { dirname, sep } from "node:path"; + +interface Config { + tracer: Tracer, + // A predicate over the trace name. A trace is forwarded to the wrapped tracer + // only if this returns true for its name. When omitted, every trace is + // forwarded. + traceFilter?: (name: string) => boolean; + // When true, the source file and line where each forwarded trace was called are + // added to its attributes as `code.filepath` / `code.lineno`. Off by default — + // capturing a call site walks a stack frame, so it is gated. + includeCallSite?: boolean; +} + +class ProxyTracer implements Tracer { + private readonly inner: Tracer; + // The name predicate, or `undefined` for "no filtering — forward all". + private readonly traceFilter: ((name: string) => boolean) | undefined; + private readonly includeCallSite: boolean; + + constructor(config: Config) { + this.inner = config.tracer; + this.traceFilter = config.traceFilter; + this.includeCallSite = config.includeCallSite ?? false; + } + + // Whether a trace with this name should be forwarded to the wrapped tracer. + // No filter → forward everything; otherwise the predicate decides. + private enabled(name: string): boolean { + return this.traceFilter === undefined || this.traceFilter(name); + } + + // Merge in the caller's source location when configured. Only called on the + // forwarding path, so the stack-walk cost is paid only for traces we keep. + private withCallSite(attributes: Attributes): Attributes { + if (!this.includeCallSite) { + return attributes; + } + const site = callSite(); + return site === undefined ? attributes : { ...attributes, ...site }; + } + + trace(name: string, attributes: Attributes, f: () => A): A { + if (!this.enabled(name)) { + return f(); + } + return this.inner.trace(name, this.withCallSite(attributes), f); + } + + traceP(name: string, attributes: Attributes, f: () => Promise): Promise { + if (!this.enabled(name)) { + return f(); + } + return this.inner.traceP(name, this.withCallSite(attributes), f); + } + + traceF(name: string, attributes: Attributes, f: Future): Future { + if (!this.enabled(name)) { + return f; + } + return this.inner.traceF(name, this.withCallSite(attributes), f); + } + + // Events are always forwarded: the wrapped tracer attaches them to the active + // span if one exists and no-ops otherwise, so an event naturally lands on its + // enclosing span when that span passed the filter, and is dropped when it did + // not (no span is active to attach to). + event(name: string, attributes?: Attributes): void { + this.inner.event(name, attributes); + } +} + +// Frames inside this directory (the tracing facade + tracers) are skipped when +// looking for the caller's location. +const TRACING_PREFIX = dirname(fileURLToPath(import.meta.url)) + sep; + +// Only need to see past the handful of tracing frames to reach the caller. +const STACK_LIMIT = 12; + +// The source location of the first stack frame outside the tracing library, as +// OpenTelemetry `code.*` attributes. Undefined if it can't be determined. +function callSite(): Attributes | undefined { + const previousLimit = Error.stackTraceLimit; + Error.stackTraceLimit = STACK_LIMIT; // only need one frame past ours; keep it cheap + const holder: { stack?: string } = {}; + Error.captureStackTrace(holder); + Error.stackTraceLimit = previousLimit; + + for (const line of holder.stack?.split("\n").slice(1) ?? []) { + const frame = parseFrame(line); + if (frame === undefined || frame.file.startsWith(TRACING_PREFIX)) { + continue; // skip non-frames and the tracing library's own frames + } + return { "code.filepath": frame.file, "code.lineno": frame.line }; + } + return undefined; +} + +// Pull "file:line:col" out of a V8 stack frame line, with or without the +// "at fn (...)" wrapper. +function parseFrame(line: string): { file: string; line: number } | undefined { + const match = line.match(/\((.+):(\d+):(\d+)\)\s*$/) ?? line.match(/at\s+(?:async\s+)?(.+):(\d+):(\d+)\s*$/); + const path = match?.[1]; + const lineno = match?.[2]; + if (path === undefined || lineno === undefined) { + return undefined; + } + const file = path.startsWith("file://") ? fileURLToPath(path) : path; + return { file, line: Number(lineno) }; +} diff --git a/core/src/tracing/simple.ts b/core/src/tracing/simple.ts new file mode 100644 index 0000000..4a0befc --- /dev/null +++ b/core/src/tracing/simple.ts @@ -0,0 +1,272 @@ +// This module allows for the issuing of tracing events which can be seen +// in a tracing explorer like https://ui.perfetto.dev/. +// +// It works by emitting the traces to stdout. You should then pipe stdout into a +// JSON file, filtering only the lines that contain trace events. +// +// It uses AsyncLocalStorage to nest traces, so there is no need to pass a token +// object around and it detects nested traces through promises. +// +// Collect events with: +// $ docker logs event-sourcing-backend | grep "\"ph\"" | jq -s '.' >events.json +export { SimpleTracer } + +import type { Tracer, Attributes } from "./index"; +import { Future } from "../future"; +import { AsyncLocalStorage } from "node:async_hooks"; + +// The currently-active trace. Nested `trace`/`traceP`/`traceF` calls read this +// to find their parent, so nesting needs no token to be threaded through code. +const store = new AsyncLocalStorage(); + +// Begin a trace as a child of the active one, or a new root when none is active. +function begin(name: string, attributes: Attributes): Trace { + const parent = store.getStore(); + return parent ? parent.startSubtrace(name, attributes) : Trace.start(name, attributes); +} + +// Run `fn` with `t` as the active trace; with `undefined`, run with no active +// trace (used to restore the parent — which may be a root — when settling). +function runIn(t: Trace | undefined, fn: () => R): R { + return t ? store.run(t, fn) : store.exit(fn); +} + +// Emits Perfetto/Chrome-event JSON to stdout, nesting traces via the +// module-level AsyncLocalStorage `store` so no token is threaded through code. +class SimpleTracer implements Tracer { + + constructor(){} + + // Trace a synchronous function. + trace(name: string, attributes: Attributes, f: () => A): A { + const t = begin(name, attributes); + try { + return store.run(t, f); + } finally { + t.end(); + } + } + + // Trace an asynchronous function. `store.run` keeps the trace active across the + // awaits inside `f` (AsyncLocalStorage propagates it), so traces created within + // `f` nest under it. After the awaited promise settles we are back in the + // caller's context, so the trace ends in the right place and a subsequent + // `traceP` becomes a sibling. + async traceP(name: string, attributes: Attributes, f: () => Promise): Promise { + const t = begin(name, attributes); + try { + return await store.run(t, f); + } finally { + t.end(); + } + } + + // Trace a Future. + // N.B. Forking here is fine. It's what `bracket` and `chainRej` do under the hood too. + traceF(name: string, attributes: Attributes, f: Future): Future { + return Future.create((reject, resolve) => { + const parent = store.getStore(); + const t = parent ? parent.startSubtrace(name, attributes) : Trace.start(name, attributes); + return runIn(t, () => + f.fork( + err => { + t.end(); + runIn(parent, () => reject(err)); + }, + val => { + t.end(); + runIn(parent, () => resolve(val)); + }, + ), + ); + }); + } + + // Record a point-in-time (instant) event on the active trace's lane. Renders + // as a marker on that track in Perfetto. No-op when no trace is active. + event(name: string, attributes: Attributes = {}): void { + const t = store.getStore(); + if (!t) { + return; + } + emit({ name, ph: PhaseInstant, ts: nowMicros(), pid: PROCESS_ID, tid: t.threadId, s: "t", args: attributes }); + } +} + + +const PROCESS_ID = 1; + +// Current time in microseconds from a monotonic, high-resolution clock. +// `process.hrtime.bigint()` returns nanoseconds, so we divide by 1000. This +// avoids the millisecond-granularity collisions of a wall-clock (Date/POSIX) +// timestamp, which collapsed nested spans to zero width. +function nowMicros(): number { + return Number(process.hrtime.bigint() / 1000n); +} + +// A pool of thread ids. A root trace acquires one for its lifetime and releases +// it on end, so concurrent requests always get distinct lanes while the total +// number of lanes stays as small as the peak concurrency (rather than growing +// once per request). Node runs JavaScript on a single thread, so no lock is +// required (unlike the Python implementation). +class ThreadIdPool { + private free: number[] = []; + private next = 0; + private named = new Set(); + + acquire(): number { + const id = this.free.pop() ?? this.next++; + if (!this.named.has(id)) { + this.named.add(id); + emitThreadName(id, `request lane ${id}`); + } + return id; + } + + release(id: number): void { + this.free.push(id); + } +} + +const threadPool = new ThreadIdPool(); + +// Trace event phases — synchronous Duration events plus Metadata. +// See https://docs.google.com/document/d/1CvAClvFfyA5R-PhYUmn5OOQtYMH4h6I0nSsKchNAySU/preview +const PhaseDurationBegin = "B" as const; // Enter a duration on this thread +const PhaseDurationEnd = "E" as const; // Leave the innermost open duration on this thread +const PhaseInstant = "i" as const; // A point-in-time event on this thread +const PhaseMetadata = "M" as const; // Metadata (thread/process names) + +type Phase = typeof PhaseDurationBegin | typeof PhaseDurationEnd | typeof PhaseInstant | typeof PhaseMetadata; + +type Args = Record; + +// A tracing event. +interface Event { + name: string; + ph: Phase; + ts: number; // microseconds + pid: number; + tid: number; + // Scope of an instant ("i") event: "g" global, "p" process, "t" thread. + s?: "g" | "p" | "t"; + args: Args; +} + +/** + * An ongoing trace, occupying a track (thread) in the trace explorer. + * + * A root trace (created with `Trace.start`) owns a thread id for its lifetime. + * Subtraces share that thread id so they nest, by stack order, within the root. + * Because Duration events must be strictly nested per thread, callers must end + * subtraces before the parent ends — which `subtrace` guarantees. + */ +class Trace { + private constructor( + readonly name: string, + readonly threadId: number, + // Only a root trace owns its thread id and releases it back to the pool on + // end. Subtraces borrow the root's thread id. + private readonly ownsThread: boolean, + readonly args: Args, + ) {} + + // Start a trace on its own thread (track). + static start(name: string, args: Args = {}): Trace { + const trace = new Trace(name, threadPool.acquire(), true, args); + trace.begin(); + return trace; + } + + // Trace a subsection of the original trace, nested on the same thread. + startSubtrace( + name: string, // trace name + args: Args = {}, // any extra information to be associated with the trace. + ): Trace { + const trace = new Trace(name, this.threadId, false, args); + trace.begin(); + return trace; + } + + end(): void { + emit({ + name: this.name, + ph: PhaseDurationEnd, + ts: nowMicros(), + pid: PROCESS_ID, + tid: this.threadId, + args: {}, + }); + if (this.ownsThread) { + threadPool.release(this.threadId); + } + } + + private begin(): void { + emit({ + name: this.name, + ph: PhaseDurationBegin, + ts: nowMicros(), + pid: PROCESS_ID, + tid: this.threadId, + args: this.args, + }); + } + + // Run `body` within a nested trace, ending the subtrace afterwards even if + // `body` throws. This is the idiomatic TypeScript equivalent of the Python + // `subtrace` async context manager. + // + // await trace.subtrace("my-step", async (sub) => { + // ... // do work here. + // }); + // + async subtrace( + name: string, + body: (trace: Trace) => Promise, + args: Args = {}, + ): Promise { + const trace = this.startSubtrace(name, args); + try { + return await body(trace); + } finally { + trace.end(); + } + } +} + +let processNamed = false; + +// Name the process once, lazily, so the explorer shows a readable label. +function ensureProcessName(): void { + if (processNamed) { + return; + } + processNamed = true; + emit({ + name: "process_name", + ph: PhaseMetadata, + ts: 0, + pid: PROCESS_ID, + tid: 0, + args: { name: "backend" }, + }); +} + +// Name a thread (lane) once, the first time it is used. +function emitThreadName(tid: number, name: string): void { + ensureProcessName(); + emit({ + name: "thread_name", + ph: PhaseMetadata, + ts: 0, + pid: PROCESS_ID, + tid, + args: { name }, + }); +} + +// Emit a trace event as a JSON line on stdout. +function emit(event: Event): void { + console.log(JSON.stringify(event)); +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6cc1e5a..f2fce2f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -17,6 +17,24 @@ importers: core: dependencies: + '@opentelemetry/api': + specifier: ^1.9.1 + version: 1.9.1 + '@opentelemetry/context-async-hooks': + specifier: ^2.7.1 + version: 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': + specifier: ^2.7.1 + version: 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace-base': + specifier: ^2.7.1 + version: 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace-node': + specifier: ^2.7.1 + version: 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': + specifier: ^1.41.1 + version: 1.41.1 '@optique/core': specifier: ^0.6.2 version: 0.6.11 @@ -938,6 +956,44 @@ packages: resolution: {integrity: sha512-oGB+UxlgWcgQkgwo8GcEGwemoTFt3FIO9ababBmaGwXIoBKZ+GTy0pP185beGg7Llih/NSHSV2XAs1lnznocSg==} engines: {node: '>= 8'} + '@opentelemetry/api@1.9.1': + resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==} + engines: {node: '>=8.0.0'} + + '@opentelemetry/context-async-hooks@2.7.1': + resolution: {integrity: sha512-OPFBYuXEn1E4ja3Y6eeA7O+ZnLBNcXTV5Cgsn1VaqBZ6hC5FnpZPLBNme1LJY8ZtF4aOujPKFoeWN4ik487KuQ==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.0.0 <1.10.0' + + '@opentelemetry/core@2.7.1': + resolution: {integrity: sha512-QAqIj32AtK6+pEVNG7EOVxHdE06RP+FM5qpiEJ4RtDcFIqKUZHYhl7/7UY5efhwmwNAg7j8QbJVBLxMerc0+gw==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.0.0 <1.10.0' + + '@opentelemetry/resources@2.7.1': + resolution: {integrity: sha512-DeT6KKolmC4e/dRQvMQ/RwlnzhaqeiFOXY5ngoOPJ07GgVVKxZOg9EcrNZb5aTzUn+iCrJldAgOfQm1O/QfPAQ==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.3.0 <1.10.0' + + '@opentelemetry/sdk-trace-base@2.7.1': + resolution: {integrity: sha512-NAYIlsF8MPUsKqJMiDQJTMPOmlbawC1Iz/omMLygZ1C9am8fTKYjTaI+OZM+WTY3t3Glo0wnOg/6/pac6RGPPw==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.3.0 <1.10.0' + + '@opentelemetry/sdk-trace-node@2.7.1': + resolution: {integrity: sha512-pCpQxU68lV+I9s9svqMyVu5iHdDDUnqUpSxqwyCU8A9ejEsSnMPCbearwsUO4yk08ZJzAIUCFuReMdVQvHrdvg==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.0.0 <1.10.0' + + '@opentelemetry/semantic-conventions@1.41.1': + resolution: {integrity: sha512-/UhIkaZgPutTFmQ7RnIJGgDXZmtEJ7Dvi86xNTFWcnRxVRNk/aotsqDJYeEvDP+FSMB2SdW+pQzNMcWP0rwuNA==} + engines: {node: '>=14'} + '@optique/core@0.6.11': resolution: {integrity: sha512-GVLFihzBA1j78NFlkU5N1Lu0jRqET0k6Z66WK8VQKG/a3cxmCInVGSKMIdQG8i6pgC8wD5OizF6Y3QMztmhAxg==} engines: {bun: '>=1.2.0', deno: '>=2.3.0', node: '>=20.0.0'} @@ -3659,6 +3715,39 @@ snapshots: '@nodelib/fs.scandir': 2.1.5 fastq: 1.20.1 + '@opentelemetry/api@1.9.1': {} + + '@opentelemetry/context-async-hooks@2.7.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + + '@opentelemetry/core@2.7.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/semantic-conventions': 1.41.1 + + '@opentelemetry/resources@2.7.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': 1.41.1 + + '@opentelemetry/sdk-trace-base@2.7.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': 1.41.1 + + '@opentelemetry/sdk-trace-node@2.7.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/context-async-hooks': 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace-base': 2.7.1(@opentelemetry/api@1.9.1) + + '@opentelemetry/semantic-conventions@1.41.1': {} + '@optique/core@0.6.11': {} '@optique/core@0.6.3': {} From 11fc941ae2e4d42598bb42760e177c968fad9d78 Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 10 Jun 2026 12:16:56 +0100 Subject: [PATCH 2/6] Lint --- core/src/tracing/index.ts | 18 ++++++++---------- core/src/tracing/opentelemetry.ts | 1 - core/src/tracing/proxy.ts | 2 +- core/src/tracing/simple.ts | 12 +++--------- 4 files changed, 12 insertions(+), 21 deletions(-) diff --git a/core/src/tracing/index.ts b/core/src/tracing/index.ts index fc882df..14ef7b4 100644 --- a/core/src/tracing/index.ts +++ b/core/src/tracing/index.ts @@ -23,23 +23,23 @@ type Name = string | (Attributes & { name: string }); // explicitly (no polyvariadic signatures, no trailing attributes). interface Tracer { /** - * Trace a synchronous call. - */ + * Trace a synchronous call. + */ trace(name: string, attributes: Attributes, f: () => A): A; /** - * Trace an asynchronous call. - */ + * Trace an asynchronous call. + */ traceP(name: string, attributes: Attributes, f: () => Promise): Promise; /** - * Trace a Future - */ + * Trace a Future + */ traceF(name: string, attributes: Attributes, f: Future): Future; /** - * Record a point-in-time event on the currently-active trace. - */ + * Record a point-in-time event on the currently-active trace. + */ event(name: string, attributes?: Attributes): void; } @@ -95,5 +95,3 @@ function named(name: Name): { name: string; attributes: Attributes } { const { name: n, ...attributes } = name; return { name: n, attributes }; } - - diff --git a/core/src/tracing/opentelemetry.ts b/core/src/tracing/opentelemetry.ts index 661aea8..e939785 100644 --- a/core/src/tracing/opentelemetry.ts +++ b/core/src/tracing/opentelemetry.ts @@ -233,4 +233,3 @@ class OpenTelemetryTracer implements Tracer { otel.getActiveSpan()?.addEvent(name, attributes); } } - diff --git a/core/src/tracing/proxy.ts b/core/src/tracing/proxy.ts index 18ba7c1..419a5c1 100644 --- a/core/src/tracing/proxy.ts +++ b/core/src/tracing/proxy.ts @@ -22,7 +22,7 @@ import { fileURLToPath } from "node:url"; import { dirname, sep } from "node:path"; interface Config { - tracer: Tracer, + tracer: Tracer; // A predicate over the trace name. A trace is forwarded to the wrapped tracer // only if this returns true for its name. When omitted, every trace is // forwarded. diff --git a/core/src/tracing/simple.ts b/core/src/tracing/simple.ts index 4a0befc..b21f207 100644 --- a/core/src/tracing/simple.ts +++ b/core/src/tracing/simple.ts @@ -9,7 +9,7 @@ // // Collect events with: // $ docker logs event-sourcing-backend | grep "\"ph\"" | jq -s '.' >events.json -export { SimpleTracer } +export { SimpleTracer }; import type { Tracer, Attributes } from "./index"; import { Future } from "../future"; @@ -34,8 +34,7 @@ function runIn(t: Trace | undefined, fn: () => R): R { // Emits Perfetto/Chrome-event JSON to stdout, nesting traces via the // module-level AsyncLocalStorage `store` so no token is threaded through code. class SimpleTracer implements Tracer { - - constructor(){} + constructor() {} // Trace a synchronous function. trace(name: string, attributes: Attributes, f: () => A): A { @@ -93,7 +92,6 @@ class SimpleTracer implements Tracer { } } - const PROCESS_ID = 1; // Current time in microseconds from a monotonic, high-resolution clock. @@ -221,11 +219,7 @@ class Trace { // ... // do work here. // }); // - async subtrace( - name: string, - body: (trace: Trace) => Promise, - args: Args = {}, - ): Promise { + async subtrace(name: string, body: (trace: Trace) => Promise, args: Args = {}): Promise { const trace = this.startSubtrace(name, args); try { return await body(trace); From d8730046b219ffdd59c5ea1a1f320a55a884513e Mon Sep 17 00:00:00 2001 From: Christopher Costa Date: Thu, 11 Jun 2026 15:00:46 +0100 Subject: [PATCH 3/6] feat(tracing): OpenTelemetryTracer.shutdown (forceFlush+shutdown), optional sampler, error status on failures --- core/src/tracing/opentelemetry.ts | 34 ++++++++++++++++++++++++++++--- 1 file changed, 31 insertions(+), 3 deletions(-) diff --git a/core/src/tracing/opentelemetry.ts b/core/src/tracing/opentelemetry.ts index e939785..6855968 100644 --- a/core/src/tracing/opentelemetry.ts +++ b/core/src/tracing/opentelemetry.ts @@ -20,6 +20,7 @@ import { Future } from "../future"; import { context, trace as otel, + SpanStatusCode, type Context, type Span, type Tracer as OtelTracer, @@ -27,7 +28,7 @@ import { type AttributeValue, } from "@opentelemetry/api"; import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; -import type { SpanProcessor, ReadableSpan } from "@opentelemetry/sdk-trace-base"; +import type { SpanProcessor, ReadableSpan, Sampler } from "@opentelemetry/sdk-trace-base"; import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; import { resourceFromAttributes } from "@opentelemetry/resources"; import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions"; @@ -154,19 +155,32 @@ interface OpenTelemetryTracerOptions { // Where finished spans go. Use `new OtlpJsonStdoutProcessor()` for the stdout // OTLP/JSON output, or a BatchSpanProcessor + OTLP exporter for a backend. spanProcessor: SpanProcessor; + // Head sampling decision. When omitted the SDK default applies (honouring the + // OTEL_TRACES_SAMPLER env), so sampling can be left to the env or the collector. + sampler?: Sampler; +} + +// Mark a span as failed: record the error and set its status to ERROR. Accepts +// any thrown/rejected value (Futures reject with an arbitrary `E`), coercing +// non-Error values to a string. +function recordError(span: Span, err: unknown): void { + span.recordException(err instanceof Error ? err : String(err)); + span.setStatus({ code: SpanStatusCode.ERROR }); } class OpenTelemetryTracer implements Tracer { private readonly tracer: OtelTracer; + private readonly provider: NodeTracerProvider; constructor(options: OpenTelemetryTracerOptions) { - const provider = new NodeTracerProvider({ + this.provider = new NodeTracerProvider({ resource: resourceFromAttributes({ [ATTR_SERVICE_NAME]: options.serviceName }), spanProcessors: [options.spanProcessor], + ...(options.sampler ? { sampler: options.sampler } : {}), }); // The AsyncLocalStorage context manager is what lets `context.active()` // return the right parent span inside `await`s and Fluture continuations. - provider.register({ contextManager: new AsyncLocalStorageContextManager().enable() }); + this.provider.register({ contextManager: new AsyncLocalStorageContextManager().enable() }); this.tracer = otel.getTracer(options.serviceName); } @@ -183,6 +197,9 @@ class OpenTelemetryTracer implements Tracer { const { span, ctx } = this.begin(name, attributes); try { return context.with(ctx, f); + } catch (err) { + recordError(span, err); + throw err; } finally { span.end(); } @@ -197,6 +214,9 @@ class OpenTelemetryTracer implements Tracer { const { span, ctx } = this.begin(name, attributes); try { return await context.with(ctx, f); + } catch (err) { + recordError(span, err); + throw err; } finally { span.end(); } @@ -215,6 +235,7 @@ class OpenTelemetryTracer implements Tracer { return context.with(ctx, () => f.fork( err => { + recordError(span, err); span.end(); context.with(parentCtx, () => reject(err)); }, @@ -232,4 +253,11 @@ class OpenTelemetryTracer implements Tracer { event(name: string, attributes?: Attributes): void { otel.getActiveSpan()?.addEvent(name, attributes); } + + // Flush any pending spans, then shut down the provider and its span processors. + // Call on graceful shutdown (SIGTERM/SIGINT) so the final batch is not dropped. + async shutdown(): Promise { + await this.provider.forceFlush(); + await this.provider.shutdown(); + } } From d80bc7cda08dbd380f1e72d1715dbf623fcea4ce Mon Sep 17 00:00:00 2001 From: Christopher Costa Date: Thu, 11 Jun 2026 15:29:01 +0100 Subject: [PATCH 4/6] ci: authenticate GitHub Packages via ~/.npmrc (pnpm 10 drops env-var auth in repo .npmrc) --- .github/workflows/ambar-core.yaml | 5 ++++- .github/workflows/ambar-task-explorer.yaml | 10 ++++++++-- .github/workflows/ambar-tasks.yaml | 5 ++++- 3 files changed, 16 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ambar-core.yaml b/.github/workflows/ambar-core.yaml index b7bb839..d24397c 100644 --- a/.github/workflows/ambar-core.yaml +++ b/.github/workflows/ambar-core.yaml @@ -26,7 +26,10 @@ jobs: version: 10 - name: Install - run: pnpm install + # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead. + run: | + echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc + pnpm install env: GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }} diff --git a/.github/workflows/ambar-task-explorer.yaml b/.github/workflows/ambar-task-explorer.yaml index f426429..10ce396 100644 --- a/.github/workflows/ambar-task-explorer.yaml +++ b/.github/workflows/ambar-task-explorer.yaml @@ -30,7 +30,10 @@ jobs: version: 10 - name: Install - run: pnpm install + # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead. + run: | + echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc + pnpm install env: GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }} @@ -74,7 +77,10 @@ jobs: version: 10 - name: Install - run: pnpm install + # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead. + run: | + echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc + pnpm install env: GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }} diff --git a/.github/workflows/ambar-tasks.yaml b/.github/workflows/ambar-tasks.yaml index b64c9bd..1ffe792 100644 --- a/.github/workflows/ambar-tasks.yaml +++ b/.github/workflows/ambar-tasks.yaml @@ -40,7 +40,10 @@ jobs: version: 10 - name: Install - run: pnpm install + # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead. + run: | + echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc + pnpm install env: GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }} From 227ae8f51358ae0cbda5a4296238c759f3ae4458 Mon Sep 17 00:00:00 2001 From: Christopher Costa Date: Thu, 11 Jun 2026 16:01:37 +0100 Subject: [PATCH 5/6] chore(core): bump @ambarltd/core to 0.1.17 (tracing module) --- core/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/package.json b/core/package.json index f02568f..1ae016c 100644 --- a/core/package.json +++ b/core/package.json @@ -1,6 +1,6 @@ { "name": "@ambarltd/core", - "version": "0.1.16", + "version": "0.1.17", "type": "module", "repository": "https://github.com/ambarltd/typescript-libs.git", "publishConfig": { From 4cfe2335806fe23d01d0f987ab4b0a7e286eb697 Mon Sep 17 00:00:00 2001 From: Christopher Costa Date: Thu, 11 Jun 2026 16:01:37 +0100 Subject: [PATCH 6/6] ci(task-explorer): authenticate GitHub Packages via ~/.npmrc in Docker build (pnpm 10) --- task-explorer/Dockerfile | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/task-explorer/Dockerfile b/task-explorer/Dockerfile index e56f75c..8196029 100644 --- a/task-explorer/Dockerfile +++ b/task-explorer/Dockerfile @@ -14,7 +14,11 @@ RUN : "${GITHUB_TOKEN:?Required build argument GITHUB_TOKEN not set}" COPY ./frontend/package.json ./package.json COPY ./.npmrc ./.npmrc -RUN --mount=type=cache,id=pnpm-store-${TARGETPLATFORM},target=/root/.local/share/pnpm/store pnpm install +# pnpm 10 ignores env-var auth in the repo .npmrc; use a literal token in ~/.npmrc (removed in-layer). +RUN --mount=type=cache,id=pnpm-store-${TARGETPLATFORM},target=/root/.local/share/pnpm/store \ + echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc && \ + pnpm install && \ + rm -f ~/.npmrc COPY ./frontend/src ./src COPY ./frontend/public ./public @@ -35,7 +39,11 @@ RUN : "${GITHUB_TOKEN:?Required build argument GITHUB_TOKEN not set}" COPY ./backend/package.json ./package.json COPY ./.npmrc ./.npmrc -RUN --mount=type=cache,id=pnpm-store-${TARGETPLATFORM},target=/root/.local/share/pnpm/store pnpm install +# pnpm 10 ignores env-var auth in the repo .npmrc; use a literal token in ~/.npmrc (removed in-layer). +RUN --mount=type=cache,id=pnpm-store-${TARGETPLATFORM},target=/root/.local/share/pnpm/store \ + echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc && \ + pnpm install && \ + rm -f ~/.npmrc COPY ./backend/src ./src COPY ./backend/tests ./tests