From 9e7d8797958bbd7b94b848ceee0c280ea54dc83a Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 2 Sep 2026 00:37:56 +0200 Subject: [PATCH] feat(effect-sdk): opt-in native Cloudflare tracer for the Workers preset Add `tracer: "native"` to `MapleCloudflareSDK.make`. Every sampled Effect span is mirrored onto `tracing.startActiveSpan` from `cloudflare:workers`, so it lands in the same trace as Cloudflare's own fetch/KV/R2/D1 spans and is exported by the Worker's ObservabilityDestination: no in-isolate buffer, no `ctx.waitUntil` flush, no ingest key in the Worker. That also makes it usable from Durable Object and Workflow isolates, where the flush is unreliable. Each mirrored span captures `AsyncLocalStorage.snapshot()` inside its callback, children open inside the parent's snapshot, and the tracer's `context` hook runs every fiber step inside the current span's snapshot, so spans opened after a yield still nest correctly and runtime spans attach under the right Effect span. Cloudflare's `isTraced` cascades into Effect's `sampled`. Cloudflare spans carry only scalar attributes, so a failed exit is recorded as `exception.type` / `exception.message` / `exception.stacktrace` / `error.type` instead of an `exception` event; non-scalar attributes, events and links stay Effect-local. The exit classification shared with the OTLP buffer tracer moves to `shared/span-exit.ts` so both paths agree on what is an error. Logs stay with Workers Logs in this mode (no OTLP logger): their Effect trace ids would never match the Cloudflare trace ids on the exported spans. `cloudflare:workers` and `node:async_hooks` are imported dynamically by a variable specifier and narrowed by type guards; when either API is absent the layer logs one notice and keeps spans Effect-local. The default mode and the api/alerting workers are unchanged. --- packages/effect-sdk/README.md | 47 ++- .../effect-sdk/src/cloudflare/index.test.ts | 28 +- packages/effect-sdk/src/cloudflare/index.ts | 24 ++ .../src/cloudflare/native-tracer.test.ts | 375 ++++++++++++++++++ .../src/cloudflare/native-tracer.ts | 254 ++++++++++++ .../effect-sdk/src/shared/flushable-tracer.ts | 78 +--- packages/effect-sdk/src/shared/span-exit.ts | 69 ++++ 7 files changed, 804 insertions(+), 71 deletions(-) create mode 100644 packages/effect-sdk/src/cloudflare/native-tracer.test.ts create mode 100644 packages/effect-sdk/src/cloudflare/native-tracer.ts create mode 100644 packages/effect-sdk/src/shared/span-exit.ts diff --git a/packages/effect-sdk/README.md b/packages/effect-sdk/README.md index e2de3d31c..0d9c6e53e 100644 --- a/packages/effect-sdk/README.md +++ b/packages/effect-sdk/README.md @@ -77,16 +77,51 @@ When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are draine ### Cloudflare-specific options -| Option | Description | -| ----------------------------- | ----------------------------------------------------------------------------------------------------------- | +| Option | Description | +| ----------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `anticipatedErrorIdentifiers` | Stable `_tag` / `Error.name` identifiers for expected 4xx failures; exported as `Ok` without an exception. A failure wrapped in an `{ error: … }` envelope is matched on the body's `_tag`, so an error decoded from an HTTP response classifies the same as the class that raised it | -| `dropSpanNames` | Span names whose prefix matches an entry are dropped before OTLP export (e.g. `"McpServer/Notifications."`) | -| `excludeLogSpans` | Skip Effect log spans in OTLP log attributes. Default `false` | -| `tracesPath` | OTLP traces path appended to `endpoint`. Default `/v1/traces` | -| `logsPath` | OTLP logs path appended to `endpoint`. Default `/v1/logs` | +| `dropSpanNames` | Span names whose prefix matches an entry are dropped before OTLP export (e.g. `"McpServer/Notifications."`) | +| `excludeLogSpans` | Skip Effect log spans in OTLP log attributes. Default `false` | +| `tracesPath` | OTLP traces path appended to `endpoint`. Default `/v1/traces` | +| `logsPath` | OTLP logs path appended to `endpoint`. Default `/v1/logs` | +| `tracer` | `"otlp"` (default) buffers spans for `flush(env)`; `"native"` mirrors them onto Cloudflare's own tracing — see [Native tracing](#native-tracing-experimental) | The same `MAPLE_ENDPOINT` / `MAPLE_INGEST_KEY` / `MAPLE_ENVIRONMENT` env vars apply, read from the Workers `env` binding. +### Native tracing (experimental) + +`tracer: "native"` hands span export to Cloudflare instead of the OTLP buffer. Every Effect span is mirrored onto `tracing.startActiveSpan` from `cloudflare:workers`, so it lands in the same trace as Cloudflare's own fetch / KV / R2 / D1 spans, and the whole trace reaches Maple through the Worker's [ObservabilityDestination](https://developers.cloudflare.com/workers/observability/exporting-opentelemetry-data/). Nothing is buffered in the isolate, no ingest key is read, and it works from Durable Object and Workflow isolates, where a `ctx.waitUntil` flush is unreliable. + +```typescript +const telemetry = MapleCloudflareSDK.make({ tracer: "native" }) +// Handler wiring is unchanged; `telemetry.flush(env)` resolves immediately in this mode. +``` + +Requirements: `compatibility_date >= 2026-07-28` (for `startActiveSpan`), the `nodejs_compat` compatibility flag (for `AsyncLocalStorage.snapshot`, which is how a span opened after a fiber yields still nests under its parent), `observability.traces.enabled = true`, and an ObservabilityDestination pointed at Maple's OTLP endpoint. When either runtime API is missing the layer logs one notice and keeps spans Effect-local — the Worker keeps running, nothing is exported. The layer builds asynchronously (it imports both modules on first build); `HttpRouter.toWebHandler` handles that. + +What is mirrored: + +| Effect | Cloudflare span | +| ----------------------------------------------------------------- | ----------------------------------------------------------------------------------------- | +| span name, nesting | span name, parent — including across `sleep`s, forks and other fiber yields | +| string / number / boolean attributes | forwarded as they are set | +| object / array / bigint attributes, events, links | Effect-local only | +| failed exit | `exception.type`, `exception.message`, `exception.stacktrace`, `error.type` (first error) | +| interrupt | `status.interrupted = true` | +| `anticipatedErrorIdentifiers` / `[ErrorReporter.ignore]` failures | no exception attributes | +| `dropSpanNames` match | no Cloudflare span; its children attach to the nearest mirrored ancestor | +| Cloudflare `isTraced = false` | the span and its descendants are unsampled — no `startActiveSpan` calls at all | + +Cloudflare spans carry no events, so a failure is recorded as attributes rather than as the OTLP `exception` event. Maple's error tracking reads both shapes. + +**Logs stay with Cloudflare.** Native mode installs no OTLP logger: Effect's default logger writes to `console`, which is Workers Logs, and the same ObservabilityDestination exports those next to the traces with Cloudflare's trace ids on them. Shipping Effect log records over OTLP would need exactly the flush and ingest key this mode removes, and every record would carry an Effect trace id that never matches the Cloudflare trace id on the exported spans. Metrics are likewise not exported in native mode. + +**Trace ids.** `Effect.currentSpan`'s `traceId` / `spanId` are independent of Cloudflare's; the ids in the exported trace are Cloudflare's. Trace context is not yet propagated to services outside Cloudflare — see Cloudflare's [known limitations](https://developers.cloudflare.com/workers/observability/traces/known-limitations/). + +**Async context.** A span runs its fibers inside the async context captured when it was opened. Two consequences: a root span opened after its fiber has already yielded (a forked background fiber with no parent span, say) attaches to whatever Cloudflare span is active at that moment, and code in the same continuation right after a span ends can still see that span as active. `HttpMiddleware.tracer` opens the request span synchronously inside the handler, which is the well-behaved case. + +Resource attributes (`serviceName`, `environment`, `attributes`) are not applied in native mode; the export carries Cloudflare's own resource attributes for the Worker. + ## Client (Browser) All configuration must be provided programmatically since browsers don't have access to environment variables. diff --git a/packages/effect-sdk/src/cloudflare/index.test.ts b/packages/effect-sdk/src/cloudflare/index.test.ts index 8078f7766..667e75c6e 100644 --- a/packages/effect-sdk/src/cloudflare/index.test.ts +++ b/packages/effect-sdk/src/cloudflare/index.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest" -import { Duration, Effect, Fiber, Layer } from "effect" +import { Duration, Effect, Fiber, Layer, Logger } from "effect" import { TestClock } from "effect/testing" import { afterEach, expect, vi } from "vitest" import { make } from "./index.js" @@ -383,4 +383,30 @@ describe("MapleCloudflareSDK.make", () => { expect(a).toBe(b) expect(Layer.isLayer(a)).toBe(true) }) + + // Native mode never touches the network: Cloudflare exports the spans. Off + // Workers (here: Node, no `cloudflare:workers`) the layer must still build + // and keep spans Effect-local rather than fail the host. + it("native mode: no fetch, flush resolves, and spans fall back to Effect-local off Workers", async () => { + const { calls, restore: r } = setupFetch() + restore = r + const notices: Array = [] + const capture = Logger.make(({ message }) => { + notices.push(Array.isArray(message) ? message.join(" ") : String(message)) + }) + const telemetry = make({ serviceName: "unit-test", tracer: "native" }) + + const span = await Effect.runPromise( + Effect.currentSpan.pipe( + Effect.withSpan("op"), + Effect.provide(telemetry.layer.pipe(Layer.provide(Logger.layer([capture])))), + ), + ) + await telemetry.flush(env) + + expect(span.name).toBe("op") + expect(calls.length).toBe(0) + expect(notices).toHaveLength(1) + expect(notices[0]).toContain("native tracing unavailable") + }) }) diff --git a/packages/effect-sdk/src/cloudflare/index.ts b/packages/effect-sdk/src/cloudflare/index.ts index 608d23d85..b718046b3 100644 --- a/packages/effect-sdk/src/cloudflare/index.ts +++ b/packages/effect-sdk/src/cloudflare/index.ts @@ -45,6 +45,7 @@ import { makeSpanBuffer, type SpanBuffer } from "../shared/flushable-tracer.js" import { makeNoOpNotice } from "../shared/no-op-notice.js" import { resolveResourceFromEnv } from "../server/resource.js" import { SDK_VERSION } from "../version.js" +import { makeNativeTracerLayer } from "./native-tracer.js" export interface Config { /** @@ -101,6 +102,19 @@ export interface Config { readonly logsPath?: string | undefined /** OTLP metrics path appended to `endpoint`. Default `/v1/metrics`. */ readonly metricsPath?: string | undefined + /** + * How spans leave the Worker. + * + * - `"otlp"` (default): spans, logs and metrics are buffered in the isolate + * and POSTed to Maple on `flush(env)`. + * - `"native"` (experimental): every span is mirrored onto Cloudflare's + * `tracing.startActiveSpan`, so it is exported by the Worker's + * ObservabilityDestination in the same trace as Cloudflare's own + * fetch/KV/R2/D1 spans. No ingest key, `flush` is a no-op, and logs are + * left to Workers Logs. Needs `compatibility_date >= 2026-07-28` and the + * `nodejs_compat` flag; when either is missing, spans stay Effect-local. + */ + readonly tracer?: "otlp" | "native" | undefined } export interface Telemetry { @@ -147,6 +161,16 @@ export const make = (config: Config = {}): Telemetry => { ] const anticipatedIdentifiers = anticipatedErrorIdentifiers.length > 0 ? new Set(anticipatedErrorIdentifiers) : undefined + + if (config.tracer === "native") { + return { + layer: makeNativeTracerLayer({ dropSpan, anticipatedErrorIdentifiers: anticipatedIdentifiers }), + // Cloudflare exports the mirrored spans itself; `flush` stays so the + // handler wiring is the same in both modes. + flush: () => Promise.resolve(), + } + } + const spans: SpanBuffer = makeSpanBuffer({ dropSpan, anticipatedErrorIdentifiers: anticipatedIdentifiers, diff --git a/packages/effect-sdk/src/cloudflare/native-tracer.test.ts b/packages/effect-sdk/src/cloudflare/native-tracer.test.ts new file mode 100644 index 000000000..511c30ae8 --- /dev/null +++ b/packages/effect-sdk/src/cloudflare/native-tracer.test.ts @@ -0,0 +1,375 @@ +import { assert, describe, it } from "@effect/vitest" +import { Data, Effect, Fiber, Layer, Logger, Option, Predicate, Tracer } from "effect" +import * as ErrorReporter from "effect/ErrorReporter" +import { expect } from "vitest" +import { + type AsyncSnapshot, + makeNativeTracer, + makeNativeTracerLayer, + type NativeSpanHandle, + type NativeTracing, + NativeTracingUnavailable, + resolveNativeTracerHost, +} from "./native-tracer.js" + +// The package typechecks without Node's globals on purpose (it also ships to +// browsers), so the real AsyncLocalStorage — the point of these tests — is +// loaded the way the tracer itself loads it: dynamically, behind a guard. +interface AsyncStore { + run(store: T, fn: () => R): R + getStore(): T | undefined +} +type AsyncHooksModule = { + readonly AsyncLocalStorage: (new () => AsyncStore) & { snapshot(): AsyncSnapshot } +} +const isAsyncHooksModule = (value: unknown): value is AsyncHooksModule => + Predicate.hasProperty(value, "AsyncLocalStorage") && Predicate.isFunction(value.AsyncLocalStorage) +const asyncHooksSpecifier = "node:async_hooks" +const loadedAsyncHooks: unknown = await import(/* @vite-ignore */ asyncHooksSpecifier) +const asyncHooks: AsyncHooksModule = isAsyncHooksModule(loadedAsyncHooks) + ? loadedAsyncHooks + : assert.fail("node:async_hooks is unavailable") +const { AsyncLocalStorage } = asyncHooks + +// A stand-in for `cloudflare:workers`' `tracing`: the active span lives in an +// AsyncLocalStorage, exactly like the runtime's async-context parenting, and +// `startActiveSpan` keeps its span active only while the callback runs. +class FakeSpan implements NativeSpanHandle { + readonly attributes: Record = {} + ended = 0 + constructor( + readonly name: string, + readonly parent: FakeSpan | undefined, + readonly isTraced: boolean, + ) {} + setAttribute(key: string, value?: boolean | number | string): void { + if (value !== undefined) this.attributes[key] = value + } + end(): void { + this.ended += 1 + } +} + +const makeFakeHost = (options: { readonly isTraced?: boolean } = {}) => { + const active = new AsyncLocalStorage() + const spans: Array = [] + const tracing: NativeTracing = { + startActiveSpan(name, callback) { + const span = new FakeSpan(name, active.getStore(), options.isTraced ?? true) + spans.push(span) + return active.run(span, () => callback(span)) + }, + } + return { + tracing, + spans, + snapshot: () => AsyncLocalStorage.snapshot(), + /** The span Cloudflare would parent a runtime-created span under right now. */ + activeSpan: () => active.getStore(), + byName: (name: string) => spans.find((span) => span.name === name), + } +} + +// Resumes the fiber from an async context with no active span — what a +// scheduler hop looks like to the runtime. +const outside = AsyncLocalStorage.snapshot() +const hop = Effect.promise(() => outside(() => new Promise((resolve) => setTimeout(resolve, 1)))) + +const withTracer = (host: ReturnType) => + Effect.provideService(Tracer.Tracer, makeNativeTracer(host)) + +class Boom extends Data.TaggedError("Boom")<{ readonly message: string }> {} +class NotFound extends Data.TaggedError("NotFound")<{}> {} +class Benign extends Data.TaggedError("Benign")<{}> { + readonly [ErrorReporter.ignore] = true +} + +describe("makeNativeTracer", () => { + it.effect("mirrors nested spans with the same parentage, ending each once", () => + Effect.gen(function* () { + const host = makeFakeHost() + yield* Effect.withSpan("child")(Effect.void).pipe(Effect.withSpan("parent"), withTracer(host)) + const parent = host.byName("parent") + const child = host.byName("child") + assert.isDefined(parent) + assert.isDefined(child) + assert.strictEqual(child.parent, parent) + assert.strictEqual(parent.ended, 1) + assert.strictEqual(child.ended, 1) + }), + ) + + it.effect("keeps parentage across a scheduler hop that lands in a foreign async context", () => + Effect.gen(function* () { + const host = makeFakeHost() + yield* Effect.gen(function* () { + yield* hop + yield* Effect.withSpan("child")(Effect.void) + }).pipe(Effect.withSpan("parent"), withTracer(host)) + assert.strictEqual(host.byName("child")?.parent, host.byName("parent")) + }), + ) + + it.effect( + "runs fiber steps inside the current span's async context, so runtime spans nest under it", + () => + Effect.gen(function* () { + const host = makeFakeHost() + const seen = yield* Effect.gen(function* () { + const before = yield* Effect.sync(() => host.activeSpan()?.name) + yield* hop + const after = yield* Effect.sync(() => host.activeSpan()?.name) + return { before, after } + }).pipe(Effect.withSpan("parent"), withTracer(host)) + assert.deepStrictEqual(seen, { before: "parent", after: "parent" }) + }), + ) + + // Within one continuation the ambient context can still be the span that + // just ended (a documented limitation), so this checks from a fresh one. + it("a fiber with no span runs in the ambient async context", async () => { + const host = makeFakeHost() + const bare = await outside(() => + Effect.runPromise(Effect.sync(() => host.activeSpan()).pipe(withTracer(host))), + ) + assert.isUndefined(bare) + }) + + it.effect("interleaved fibers each stay under their own span", () => + Effect.gen(function* () { + const host = makeFakeHost() + const work = (label: string) => + Effect.gen(function* () { + yield* hop + const active = yield* Effect.sync(() => host.activeSpan()?.name) + yield* Effect.withSpan(`${label}.child`)(Effect.void) + return active + }).pipe(Effect.withSpan(label)) + const [a, b] = yield* Effect.all([Effect.forkChild(work("a")), Effect.forkChild(work("b"))]).pipe( + Effect.flatMap(([fa, fb]) => Effect.all([Fiber.join(fa), Fiber.join(fb)])), + withTracer(host), + ) + assert.strictEqual(a, "a") + assert.strictEqual(b, "b") + assert.strictEqual(host.byName("a.child")?.parent, host.byName("a")) + assert.strictEqual(host.byName("b.child")?.parent, host.byName("b")) + }), + ) + + it.effect("forwards scalar attributes and keeps the rest Effect-local", () => + Effect.gen(function* () { + const host = makeFakeHost() + const effectAttributes = yield* Effect.gen(function* () { + yield* Effect.annotateCurrentSpan({ + "a.string": "x", + "a.number": 1, + "a.boolean": true, + "a.object": { nested: 1 }, + "a.array": [1, 2], + "a.bigint": 1n, + }) + const span = yield* Effect.currentSpan + return new Map(span.attributes) + }).pipe(Effect.withSpan("op"), withTracer(host)) + assert.deepStrictEqual(host.byName("op")?.attributes, { + "a.string": "x", + "a.number": 1, + "a.boolean": true, + }) + assert.strictEqual(effectAttributes.size, 6) + assert.deepStrictEqual(effectAttributes.get("a.object"), { nested: 1 }) + }), + ) + + it.effect("mirrors a failure as exception.* and error.type attributes", () => + Effect.gen(function* () { + const host = makeFakeHost() + yield* Effect.fail(new Boom({ message: "boom" })).pipe( + Effect.withSpan("op"), + withTracer(host), + Effect.exit, + ) + const attributes = host.byName("op")?.attributes ?? {} + assert.strictEqual(attributes["exception.type"], "Boom") + assert.strictEqual(attributes["exception.message"], "boom") + assert.strictEqual(attributes["error.type"], "Boom") + expect(attributes["exception.stacktrace"]).toEqual(expect.stringContaining("boom")) + assert.strictEqual(host.byName("op")?.ended, 1) + }), + ) + + it.effect("mirrors a defect the same way", () => + Effect.gen(function* () { + const host = makeFakeHost() + yield* Effect.die(new TypeError("unexpected")).pipe( + Effect.withSpan("op"), + withTracer(host), + Effect.exit, + ) + const attributes = host.byName("op")?.attributes ?? {} + assert.strictEqual(attributes["exception.type"], "TypeError") + assert.strictEqual(attributes["exception.message"], "unexpected") + }), + ) + + it.effect("sets no exception attributes on success", () => + Effect.gen(function* () { + const host = makeFakeHost() + yield* Effect.void.pipe(Effect.withSpan("op"), withTracer(host)) + assert.deepStrictEqual(host.byName("op")?.attributes, {}) + }), + ) + + it.effect("flags an interrupt instead of recording an exception", () => + Effect.gen(function* () { + const host = makeFakeHost() + yield* Effect.interrupt.pipe(Effect.withSpan("op"), withTracer(host), Effect.exit) + assert.deepStrictEqual(host.byName("op")?.attributes, { "status.interrupted": true }) + }), + ) + + it.effect("anticipated and ignored failures set no exception attributes", () => + Effect.gen(function* () { + const host = makeFakeHost() + const tracer = makeNativeTracer(host, { anticipatedErrorIdentifiers: new Set(["NotFound"]) }) + const provide = Effect.provideService(Tracer.Tracer, tracer) + yield* Effect.fail(new NotFound()).pipe(Effect.withSpan("anticipated"), provide, Effect.exit) + yield* Effect.fail(new Benign()).pipe(Effect.withSpan("ignored"), provide, Effect.exit) + assert.deepStrictEqual(host.byName("anticipated")?.attributes, {}) + assert.deepStrictEqual(host.byName("ignored")?.attributes, {}) + }), + ) + + it.effect("cascades Cloudflare's isTraced=false into Effect's sampled and opens no descendants", () => + Effect.gen(function* () { + const host = makeFakeHost({ isTraced: false }) + const sampled = yield* Effect.gen(function* () { + yield* Effect.withSpan("child")(Effect.void) + const span = yield* Effect.currentSpan + return span.sampled + }).pipe(Effect.withSpan("parent"), withTracer(host)) + assert.isFalse(sampled) + assert.deepStrictEqual( + host.spans.map((span) => span.name), + ["parent"], + ) + }), + ) + + it.effect( + "a dropped span name stays Effect-local and its children attach to the nearest mirrored ancestor", + () => + Effect.gen(function* () { + const host = makeFakeHost() + const tracer = makeNativeTracer(host, { dropSpan: (name) => name.startsWith("noise.") }) + yield* Effect.withSpan("child")(Effect.void).pipe( + Effect.withSpan("noise.notification"), + Effect.withSpan("parent"), + Effect.provideService(Tracer.Tracer, tracer), + ) + assert.deepStrictEqual( + host.spans.map((span) => span.name), + ["parent", "child"], + ) + assert.strictEqual(host.byName("child")?.parent, host.byName("parent")) + }), + ) + + it.effect("Effect span ids stay independent of the mirrored span", () => + Effect.gen(function* () { + const host = makeFakeHost() + const ids = yield* Effect.currentSpan.pipe( + Effect.map((span) => ({ traceId: span.traceId, spanId: span.spanId, parent: span.parent })), + Effect.withSpan("op"), + withTracer(host), + ) + expect(ids.traceId).toMatch(/^[0-9a-f]{32}$/) + expect(ids.spanId).toMatch(/^[0-9a-f]{16}$/) + assert.isTrue(Option.isNone(ids.parent)) + }), + ) +}) + +describe("resolveNativeTracerHost", () => { + it.effect( + "resolves when cloudflare:workers exposes startActiveSpan and node:async_hooks a snapshot", + () => + Effect.gen(function* () { + const { tracing } = makeFakeHost() + const host = yield* resolveNativeTracerHost((specifier) => + Promise.resolve(specifier === "cloudflare:workers" ? { tracing } : asyncHooks), + ) + assert.strictEqual(host.tracing, tracing) + assert.strictEqual( + host.snapshot()(() => 42), + 42, + ) + }), + ) + + it.effect("fails with a compatibility-date hint when tracing has no startActiveSpan", () => + Effect.gen(function* () { + const error = yield* resolveNativeTracerHost((specifier) => + Promise.resolve( + specifier === "cloudflare:workers" + ? { tracing: { enterSpan: () => undefined } } + : asyncHooks, + ), + ).pipe(Effect.flip) + assert.instanceOf(error, NativeTracingUnavailable) + expect(error.message).toContain("compatibility_date") + }), + ) + + it.effect("fails with a nodejs_compat hint when AsyncLocalStorage.snapshot is missing", () => + Effect.gen(function* () { + const { tracing } = makeFakeHost() + const error = yield* resolveNativeTracerHost((specifier) => + Promise.resolve(specifier === "cloudflare:workers" ? { tracing } : {}), + ).pipe(Effect.flip) + expect(error.message).toContain("nodejs_compat") + }), + ) + + it.effect("fails when the module cannot be imported at all", () => + Effect.gen(function* () { + const error = yield* resolveNativeTracerHost(() => + Promise.reject(new Error("No such module")), + ).pipe(Effect.flip) + expect(error.message).toContain("cloudflare:workers") + }), + ) +}) + +describe("makeNativeTracerLayer", () => { + it.effect("falls back to Effect-local spans, with one notice, when the host is unavailable", () => + Effect.gen(function* () { + const notices: Array = [] + const capture = Logger.make(({ message }) => { + notices.push(Array.isArray(message) ? message.join(" ") : String(message)) + }) + const layer = makeNativeTracerLayer( + {}, + Effect.fail(new NativeTracingUnavailable({ message: "no cloudflare:workers here" })), + ).pipe(Layer.provide(Logger.layer([capture]))) + const span = yield* Effect.currentSpan.pipe(Effect.withSpan("op"), Effect.provide(layer)) + assert.strictEqual(span.name, "op") + assert.isTrue(span.sampled) + assert.strictEqual(notices.length, 1) + expect(notices[0]).toContain("native tracing unavailable") + expect(notices[0]).toContain("no cloudflare:workers here") + }), + ) + + it.effect("installs the mirroring tracer when the host resolves", () => + Effect.gen(function* () { + const host = makeFakeHost() + const layer = makeNativeTracerLayer({}, Effect.succeed(host)) + yield* Effect.withSpan("child")(Effect.void).pipe( + Effect.withSpan("parent"), + Effect.provide(layer), + ) + assert.strictEqual(host.byName("child")?.parent, host.byName("parent")) + }), + ) +}) diff --git a/packages/effect-sdk/src/cloudflare/native-tracer.ts b/packages/effect-sdk/src/cloudflare/native-tracer.ts new file mode 100644 index 000000000..1143da902 --- /dev/null +++ b/packages/effect-sdk/src/cloudflare/native-tracer.ts @@ -0,0 +1,254 @@ +// Cloudflare-native tracer — the opt-in `tracer: "native"` mode of the Workers +// preset. +// +// Every sampled Effect span is mirrored onto `tracing.startActiveSpan` from +// `cloudflare:workers`, so it lands in the same trace as Cloudflare's own +// auto-instrumented spans (fetch, KV, R2, D1, …) and is exported by the +// customer's ObservabilityDestination. Nothing is buffered, nothing is +// flushed, no ingest key is involved, and it works from Durable Object and +// Workflow isolates where a `ctx.waitUntil` flush is unreliable. +// +// Cloudflare parents a new span under whatever span is active on the JS +// async context, and `startActiveSpan` keeps its span active only while its +// callback runs. Effect fibers hop across async contexts on every yield, so +// left alone a child opened after a `sleep` would attach to the request's +// root span. Each mirrored span therefore captures `AsyncLocalStorage.snapshot()` +// from inside its callback, children are opened inside their parent's +// snapshot, and the tracer's `context` hook runs every fiber step inside the +// current span's snapshot — which is also what puts the runtime's own +// fetch/KV/R2/D1 spans under the right Effect span. +// +// Cloudflare spans carry only scalar attributes: no events, links, or status. +// Scalars are forwarded as they are set; everything else stays on the Effect +// span. A failed exit is mirrored as `exception.type` / `exception.message` / +// `exception.stacktrace` / `error.type` attributes in place of the OTLP +// `exception` event. Effect trace and span ids are independent of +// Cloudflare's — `Effect.currentSpan` keeps working, but its ids are not the +// ones in the exported trace. + +import { Effect, Layer, Option, Predicate, Schema, Tracer } from "effect" +import { classifySpanExit } from "../shared/span-exit.js" + +/** + * Structural view of the span `tracing.startActiveSpan` hands its callback. + * Typed here rather than imported: `@cloudflare/workers-types` predates + * `startActiveSpan`, and the SDK must not require Workers types to build. + */ +export interface NativeSpanHandle { + /** Cloudflare's head-sampling decision for this invocation. */ + readonly isTraced: boolean + setAttribute(key: string, value?: boolean | number | string): void + end(): void +} + +export interface NativeTracing { + startActiveSpan(name: string, callback: (span: NativeSpanHandle) => T): T +} + +/** What `AsyncLocalStorage.snapshot()` returns: runs a thunk inside the captured async context. */ +export type AsyncSnapshot = (fn: () => T) => T + +export interface NativeTracerHost { + readonly tracing: NativeTracing + readonly snapshot: () => AsyncSnapshot +} + +export interface NativeTracerOptions { + /** Same contract as the OTLP preset: a matching name is kept Effect-local, children attach to the nearest mirrored ancestor. */ + readonly dropSpan?: ((name: string) => boolean) | undefined + /** Same contract as the OTLP preset: a failure made entirely of these gets no `exception.*` attributes. */ + readonly anticipatedErrorIdentifiers?: ReadonlySet | undefined +} + +/** Raised while resolving the host APIs; the layer turns it into the Effect-local fallback. */ +export class NativeTracingUnavailable extends Schema.TaggedError()( + "@maple-dev/effect-sdk/cloudflare/NativeTracingUnavailable", + { + message: Schema.String, + cause: Schema.optionalKey(Schema.Defect()), + }, +) {} + +type SpanOptions = Parameters[0] + +const ATTR_EXCEPTION_TYPE = "exception.type" +const ATTR_EXCEPTION_MESSAGE = "exception.message" +const ATTR_EXCEPTION_STACKTRACE = "exception.stacktrace" +const ATTR_ERROR_TYPE = "error.type" +const ATTR_STATUS_INTERRUPTED = "status.interrupted" + +const isScalar = (value: unknown): value is boolean | number | string => + Predicate.isString(value) || Predicate.isNumber(value) || Predicate.isBoolean(value) + +class MirroredSpan extends Tracer.NativeSpan { + /** Async context to run this span's fibers in: the parent's, or `undefined` for the ambient one. */ + readonly runIn: AsyncSnapshot | undefined + readonly handle: NativeSpanHandle | undefined + readonly #anticipated: ReadonlySet | undefined + + constructor( + options: SpanOptions, + runIn: AsyncSnapshot | undefined, + handle: NativeSpanHandle | undefined, + sampled: boolean, + anticipated: ReadonlySet | undefined, + ) { + super({ ...options, sampled }) + this.runIn = runIn + this.handle = handle + this.#anticipated = anticipated + } + + override attribute(key: string, value: unknown): void { + super.attribute(key, value) + if (this.handle !== undefined && isScalar(value)) this.handle.setAttribute(key, value) + } + + override end(endTime: bigint, exit: Parameters[1]): void { + super.end(endTime, exit) + const handle = this.handle + if (handle === undefined) return + const outcome = classifySpanExit(exit, this.#anticipated) + if (outcome._tag === "Interrupted") { + handle.setAttribute(ATTR_STATUS_INTERRUPTED, true) + } else if (outcome._tag === "Failed") { + // One scalar per key: the first error is the one Maple fingerprints on, + // exactly as the OTLP path's first `exception` event is. + const first = outcome.errors[0] + if (first !== undefined) { + handle.setAttribute(ATTR_EXCEPTION_TYPE, first.name) + handle.setAttribute(ATTR_EXCEPTION_MESSAGE, first.message) + handle.setAttribute(ATTR_EXCEPTION_STACKTRACE, first.stack ?? "No stack trace available") + handle.setAttribute(ATTR_ERROR_TYPE, first.name) + } + } + handle.end() + } +} + +// The nearest mirrored ancestor decides the async context. The walk stops at +// an `ExternalSpan` (a propagated parent has no Cloudflare span of its own). +const runInFor = (span: Tracer.AnySpan | undefined): AsyncSnapshot | undefined => { + let current = span + while (current !== undefined && current._tag === "Span") { + if (current instanceof MirroredSpan) return current.runIn + current = Option.getOrUndefined(current.parent) + } + return undefined +} + +export const makeNativeTracer = ( + host: NativeTracerHost, + options: NativeTracerOptions = {}, +): Tracer.Tracer => { + const { tracing, snapshot } = host + const dropSpan = options.dropSpan + const anticipated = options.anticipatedErrorIdentifiers + + return Tracer.make({ + span(spanOptions) { + const parentRun = spanOptions.root + ? undefined + : runInFor(Option.getOrUndefined(spanOptions.parent)) + if (!spanOptions.sampled) { + return new MirroredSpan(spanOptions, parentRun, undefined, false, anticipated) + } + if (dropSpan !== undefined && dropSpan(spanOptions.name)) { + return new MirroredSpan(spanOptions, parentRun, undefined, true, anticipated) + } + // Snapshot from inside the callback: that is the only frame in which + // the new Cloudflare span is active. + const open = () => + tracing.startActiveSpan( + spanOptions.name, + (handle) => + new MirroredSpan(spanOptions, snapshot(), handle, handle.isTraced, anticipated), + ) + return parentRun === undefined ? open() : parentRun(open) + }, + context(primitive, fiber) { + const run = runInFor(fiber.currentSpan) + return run === undefined + ? primitive["~effect/Effect/evaluate"](fiber) + : run(() => primitive["~effect/Effect/evaluate"](fiber)) + }, + }) +} + +// Host resolution +// +// Both modules are imported dynamically, and by a non-literal specifier, so +// neither the SDK bundle nor a Worker on an older compatibility date (or one +// without `nodejs_compat`) fails at module load. A missing API degrades to +// Effect-local spans through `NativeTracingUnavailable`. + +const isNativeTracing = (value: unknown): value is NativeTracing => + Predicate.hasProperty(value, "startActiveSpan") && Predicate.isFunction(value.startActiveSpan) + +const isAsyncHooks = ( + value: unknown, +): value is { readonly AsyncLocalStorage: { readonly snapshot: () => AsyncSnapshot } } => + Predicate.hasProperty(value, "AsyncLocalStorage") && + Predicate.hasProperty(value.AsyncLocalStorage, "snapshot") && + Predicate.isFunction(value.AsyncLocalStorage.snapshot) + +/** A module namespace: named exports whose shapes are checked by the guards above. */ +export interface ModuleNamespace { + readonly [name: string]: unknown +} +export type ModuleImporter = (specifier: string) => Promise + +const importSpecifier: ModuleImporter = (specifier) => import(/* @vite-ignore */ specifier) + +export const resolveNativeTracerHost = ( + importModule: ModuleImporter, +): Effect.Effect => + Effect.gen(function* () { + const load = (specifier: string) => + Effect.tryPromise({ + try: () => importModule(specifier), + catch: (cause) => + new NativeTracingUnavailable({ message: `${specifier} could not be imported`, cause }), + }) + const workers = yield* load("cloudflare:workers") + const tracing = Predicate.hasProperty(workers, "tracing") ? workers.tracing : undefined + if (!isNativeTracing(tracing)) { + return yield* new NativeTracingUnavailable({ + message: + "cloudflare:workers exposes no `tracing.startActiveSpan` — it needs compatibility_date >= 2026-07-28", + }) + } + const asyncHooks = yield* load("node:async_hooks") + if (!isAsyncHooks(asyncHooks)) { + return yield* new NativeTracingUnavailable({ + message: + "node:async_hooks exposes no `AsyncLocalStorage.snapshot` — enable the nodejs_compat compatibility flag", + }) + } + return { tracing, snapshot: () => asyncHooks.AsyncLocalStorage.snapshot() } + }) + +const effectLocalTracer = Tracer.make({ span: (options) => new Tracer.NativeSpan(options) }) + +/** + * Tracer layer for native mode. Builds asynchronously (the host modules are + * imported on first build); when the host APIs are absent it logs one notice + * and installs Effect-local spans instead. + */ +export const makeNativeTracerLayer = ( + options: NativeTracerOptions, + host: Effect.Effect = resolveNativeTracerHost( + importSpecifier, + ), +): Layer.Layer => + Layer.effect( + Tracer.Tracer, + host.pipe( + Effect.map((resolved) => makeNativeTracer(resolved, options)), + Effect.catchTag("@maple-dev/effect-sdk/cloudflare/NativeTracingUnavailable", (error) => + Effect.logInfo( + `[MapleCloudflareSDK] native tracing unavailable — spans stay Effect-local (${error.message})`, + ).pipe(Effect.as(effectLocalTracer)), + ), + ), + ) diff --git a/packages/effect-sdk/src/shared/flushable-tracer.ts b/packages/effect-sdk/src/shared/flushable-tracer.ts index d27e8e9dc..ec5e787d7 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.ts @@ -4,10 +4,10 @@ // resource, and headers are NOT baked in here — the caller (the Cloudflare, // server, or client flushable preset) resolves them and POSTs the drained // buffer on `flush`, so the layer itself can be constructed without I/O. -import { Cause, Context, Exit, Layer, Option, Predicate, Tracer } from "effect" -import * as ErrorReporter from "effect/ErrorReporter" +import { Cause, Context, Exit, Layer, Option, Tracer } from "effect" import * as OtlpResource from "effect/unstable/observability/OtlpResource" import type { ExtractTag } from "effect/Types" +import { classifySpanExit, type SpanOutcome } from "./span-exit.js" export interface CaptureExceptionOptions { /** Span name. Default `"exception"`. */ @@ -71,25 +71,6 @@ export interface SpanBufferOptions { readonly anticipatedErrorTags?: ReadonlySet | undefined } -// Errors carrying Effect's `[ErrorReporter.ignore]` flag are benign by design — -// Effect's own "don't report this failure" signal. The canonical case is -// `HttpServerError { reason: RouteNotFound }` (unmatched routes → 404), which -// would otherwise surface as an Error-status span. We key off the annotation -// rather than concrete error tags so the check stays robust and HTTP-agnostic; -// genuine failures (400 parse errors, 500s) keep `ignore = false` and trace. -const isIgnoredFailure = (error: unknown): boolean => - Predicate.hasProperty(error, ErrorReporter.ignore) && error[ErrorReporter.ignore] === true - -const isIgnoredSpan = (span: SpanImpl): boolean => { - const status = span.status - if (status._tag !== "Ended") return false - const exit = status.exit - if (exit._tag !== "Failure") return false - if (exit.cause.reasons.some(Cause.isDieReason)) return false - const failures = exit.cause.reasons.filter(Cause.isFailReason) - return failures.length > 0 && failures.every((reason) => isIgnoredFailure(reason.error)) -} - export const makeSpanBuffer = (options: SpanBufferOptions = {}): SpanBuffer => { let buffer: Array = [] let disabled = false @@ -100,9 +81,13 @@ export const makeSpanBuffer = (options: SpanBufferOptions = {}): SpanBuffer => { if (disabled) return if (!span.sampled) return if (dropSpan !== undefined && dropSpan(span.name)) return - if (isIgnoredSpan(span)) return + if (span.status._tag !== "Ended") return + const outcome = classifySpanExit(span.status.exit, anticipatedErrorIdentifiers) + // Benign by Effect's own reckoning (`[ErrorReporter.ignore]`, e.g. a + // RouteNotFound 404): never exported, unlike the `Anticipated` case below. + if (outcome._tag === "Ignored") return if (buffer.length >= MAX_BUFFER) return - buffer.push(makeOtlpSpan(span, anticipatedErrorIdentifiers)) + buffer.push(makeOtlpSpan(span, outcome)) } const tracer = Tracer.make({ @@ -210,40 +195,7 @@ const generateId = (len: number): string => { return result } -// A failure is "anticipated" when its `_tag` is in the configured set. A span -// whose failure is caused *entirely* by anticipated errors (no defects/Die) -// records OTLP status `Ok` and emits no `exception` event. -const failureIdentifier = (error: unknown): string | undefined => { - if (Predicate.hasProperty(error, "_tag") && typeof error._tag === "string") return error._tag - if (Predicate.hasProperty(error, "name") && typeof error.name === "string") return error.name - // An error that crossed an HTTP boundary arrives as a decoded *body*, not as - // the class that raised it. An API that wraps its bodies in `{ error: … }` — - // a common envelope convention — therefore hands the failure channel a plain - // object with no identifier of its own, and every identifier a caller - // configured goes unmatched: expected 4xx answers record as `Error` spans - // whose entire message is the JSON-stringified envelope. Unwrap one level, and - // only for the body's own tag. - const body = Predicate.hasProperty(error, "error") ? error.error : undefined - if (Predicate.hasProperty(body, "_tag") && typeof body._tag === "string") return body._tag - return undefined -} - -const isAnticipatedFailure = (error: unknown, identifiers: ReadonlySet): boolean => { - const identifier = failureIdentifier(error) - return identifier !== undefined && identifiers.has(identifier) -} - -const isFullyAnticipated = ( - cause: Cause.Cause, - identifiers: ReadonlySet | undefined, -): boolean => { - if (identifiers === undefined || identifiers.size === 0) return false - if (cause.reasons.some(Cause.isDieReason)) return false - const failErrors = cause.reasons.filter(Cause.isFailReason).map((reason) => reason.error) - return failErrors.length > 0 && failErrors.every((error) => isAnticipatedFailure(error, identifiers)) -} - -const makeOtlpSpan = (self: SpanImpl, anticipatedErrorIdentifiers?: ReadonlySet): OtlpSpan => { +const makeOtlpSpan = (self: SpanImpl, outcome: SpanOutcome): OtlpSpan => { const status = self.status as ExtractTag const attributes = OtlpResource.entriesToAttributes(self.attributes.entries()) const events = self.events.map(([name, startTime, attrs]) => ({ @@ -254,20 +206,18 @@ const makeOtlpSpan = (self: SpanImpl, anticipatedErrorIdentifiers?: ReadonlySet< })) let otelStatus: Status - if (status.exit._tag === "Success") { - otelStatus = constOtelStatusSuccess - } else if (Cause.hasInterruptsOnly(status.exit.cause)) { + if (outcome._tag === "Interrupted") { otelStatus = { code: StatusCode.Ok, message: "Interrupted" } attributes.push( { key: "span.label", value: { stringValue: "⚠︎ Interrupted" } }, { key: "status.interrupted", value: { boolValue: true } }, ) - } else if (isFullyAnticipated(status.exit.cause, anticipatedErrorIdentifiers)) { - // Expected business outcome (4xx). Keep the span (latency / status code - // stay visible) but don't flag it as an error or fingerprint it. + } else if (outcome._tag !== "Failed") { + // Success, or an expected business outcome (4xx): keep the span (latency / + // status code stay visible) but don't flag it as an error or fingerprint it. otelStatus = constOtelStatusSuccess } else { - const errors = Cause.prettyErrors(status.exit.cause) + const errors = outcome.errors otelStatus = { code: StatusCode.Error } const firstError = errors[0] if (firstError) { diff --git a/packages/effect-sdk/src/shared/span-exit.ts b/packages/effect-sdk/src/shared/span-exit.ts new file mode 100644 index 000000000..84729498b --- /dev/null +++ b/packages/effect-sdk/src/shared/span-exit.ts @@ -0,0 +1,69 @@ +// How a span ended, as one value. +// +// The OTLP buffer tracer and the Cloudflare-native tracer must agree on what +// counts as an error: Maple's error tracking reads the verdict either as an +// `exception` event (OTLP) or as `exception.*` attributes (native). One +// classification keeps the two from drifting. +import { Cause, type Exit, Predicate } from "effect" +import * as ErrorReporter from "effect/ErrorReporter" + +export type SpanOutcome = + | { readonly _tag: "Success" } + /** Interrupt-only cause — not an error, but worth flagging. */ + | { readonly _tag: "Interrupted" } + /** + * Every failure carries `[ErrorReporter.ignore]`, Effect's own "don't report + * this" signal (the canonical case is `HttpServerError` / `RouteNotFound`). + */ + | { readonly _tag: "Ignored" } + /** Every failure is in the caller's `anticipatedErrorIdentifiers` (an expected 4xx). */ + | { readonly _tag: "Anticipated" } + | { readonly _tag: "Failed"; readonly errors: ReadonlyArray } + +const isIgnoredFailure = (error: unknown): boolean => + Predicate.hasProperty(error, ErrorReporter.ignore) && error[ErrorReporter.ignore] === true + +// An error that crossed an HTTP boundary arrives as a decoded *body*, not as +// the class that raised it. An API that wraps its bodies in `{ error: … }` +// would otherwise leave every configured identifier unmatched, so unwrap one +// level, and only for the body's own tag. +const failureIdentifier = (error: unknown): string | undefined => { + if (Predicate.hasProperty(error, "_tag") && typeof error._tag === "string") return error._tag + if (Predicate.hasProperty(error, "name") && typeof error.name === "string") return error.name + const body = Predicate.hasProperty(error, "error") ? error.error : undefined + if (Predicate.hasProperty(body, "_tag") && typeof body._tag === "string") return body._tag + return undefined +} + +const isAnticipatedFailure = (error: unknown, identifiers: ReadonlySet): boolean => { + const identifier = failureIdentifier(error) + return identifier !== undefined && identifiers.has(identifier) +} + +/** + * Classify a span's exit. A cause that mixes an ignored or anticipated failure + * with a defect (`Die`) is a real failure — only causes made entirely of + * benign failures are downgraded. + */ +export const classifySpanExit = ( + exit: Exit.Exit, + anticipatedErrorIdentifiers: ReadonlySet | undefined, +): SpanOutcome => { + if (exit._tag === "Success") return { _tag: "Success" } + const cause = exit.cause + if (Cause.hasInterruptsOnly(cause)) return { _tag: "Interrupted" } + if (!cause.reasons.some(Cause.isDieReason)) { + const failures = cause.reasons.filter(Cause.isFailReason).map((reason) => reason.error) + if (failures.length > 0) { + if (failures.every(isIgnoredFailure)) return { _tag: "Ignored" } + if ( + anticipatedErrorIdentifiers !== undefined && + anticipatedErrorIdentifiers.size > 0 && + failures.every((error) => isAnticipatedFailure(error, anticipatedErrorIdentifiers)) + ) { + return { _tag: "Anticipated" } + } + } + } + return { _tag: "Failed", errors: Cause.prettyErrors(cause) } +}