1import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
2import * as Cloudflare from "@/Cloudflare/index.ts";
3import {
4 CloudflareTelemetryCompatibilityError,
5 MIN_CLOUDFLARE_TRACING_DATE,
6} from "@/Cloudflare/Workers/Telemetry.ts";
7import { resolveObservability } from "@/Cloudflare/Workers/WorkerAsyncBindings.ts";
8import type { Worker } from "@/Cloudflare/Workers/Worker.ts";
9import type { ResourceBinding } from "@/Resource.ts";
10import * as Test from "@/Test/Alchemy";
11import * as workers from "@distilled.cloud/cloudflare/workers";
12import { describe, expect, test as unit } from "alchemy-test";
13import * as ConfigProvider from "effect/ConfigProvider";
14import * as Effect from "effect/Effect";
15import * as HttpClient from "effect/http/HttpClient";
16import { MinimumLogLevel } from "effect/References";
17import * as Schedule from "effect/Schedule";
18import NativeTracingWorker, {
19 makeTracedWorker,
20} from "./fixtures/native-tracing/worker.ts";
21import { expectUrlContains } from "../Utils/Http.ts";
22import { waitForWorkerToBeDeleted } from "../Utils/Worker.ts";
23
24const { test } = Test.make({ providers: Cloudflare.providers() });
25
26const logLevel = Effect.provideService(
27 MinimumLogLevel,
28 process.env.DEBUG ? "Debug" : "Info",
29);
30
31/**
32 * A span row Workers Observability ingested: Cloudflare's own
33 * `traceId` / `spanId` / `parentSpanId` (so the tree the platform recorded
34 * can be checked directly) plus the `effect.exit` attribute the tracer
35 * forwards on span end.
36 */
37interface SpanRow
38 extends workers.ObservabilitySharedQueriesGetResponseEventsEventsItemMetadata {
39 exit: unknown;
40}
41
42const exitOf = (source: unknown): unknown => {
43 if (typeof source !== "object" || source === null) return undefined;
44 const effect = (source as Record<string, unknown>).effect;
45 return typeof effect === "object" && effect !== null
46 ? (effect as Record<string, unknown>).exit
47 : undefined;
48};
49
50const querySpans = (accountId: string, service: string) =>
51 // The timeframe is re-evaluated per poll: rows are keyed by ingestion
52 // time, so a window fixed at the first attempt never sees late arrivals.
53 Effect.suspend(() =>
54 workers.queryObservabilityTelemetry({
55 accountId,
56 queryId: "adhoc-native-tracing",
57 view: "events",
58 timeframe: { from: Date.now() - 15 * 60 * 1000, to: Date.now() },
59 limit: 500,
60 parameters: {
61 filters: [
62 {
63 key: "$metadata.service",
64 operation: "eq",
65 type: "string",
66 value: service,
67 },
68 { key: "$metadata.spanName", operation: "exists", type: "string" },
69 ],
70 },
71 }),
72 ).pipe(
73 Effect.map((response) =>
74 (response.events?.events ?? []).map((event): SpanRow => ({
75 ...event.metadata,
76 exit: exitOf(event.source),
77 })),
78 ),
79 );
80
81/** The spans of one request: the root matched by name + `request.id`, and every row sharing its trace. */
82const traceOf = (spans: SpanRow[], root: string, id: string) => {
83 const rootSpan = spans.find((s) => s.spanName === root && s.requestId === id);
84 return rootSpan === undefined
85 ? undefined
86 : {
87 root: rootSpan,
88 spans: spans.filter((s) => s.traceId === rootSpan.traceId),
89 /** Exactly one span of this name in the trace. */
90 span: (name: string) => {
91 const matches = spans.filter(
92 (s) => s.traceId === rootSpan.traceId && s.spanName === name,
93 );
94 expect(matches).toHaveLength(1);
95 return matches[0]!;
96 },
97 };
98};
99
100type Bindings = ReadonlyArray<ResourceBinding<Worker["Binding"]>>;
101
102const tracesBind = (
103 traces: NonNullable<Worker["Binding"]["observability"]>["traces"],
104): Bindings =>
105 [
106 { sid: "Cloudflare.Telemetry", data: { observability: { traces } } },
107 ] as unknown as Bindings;
108
109describe(
110 "resolveObservability",
111 {
112 tags: [
113 "unit",
114 "provider:cloudflare",
115 "provider:cloudflare:worker",
116 "local",
117 ],
118 },
119 () => {
120 unit("omitted props + bound traces keep default logs", () => {
121 const resolved = resolveObservability({}, tracesBind({ enabled: true }));
122 expect(resolved.enabled).toBe(true);
123 expect(resolved.logs?.enabled).toBe(true);
124 expect(resolved.logs?.invocationLogs).toBe(true);
125 expect(resolved.traces?.enabled).toBe(true);
126 });
127
128 unit("explicit traces.enabled false wins over the bind", () => {
129 const resolved = resolveObservability(
130 { observability: { enabled: true, traces: { enabled: false } } },
131 tracesBind({ enabled: true, persist: true }),
132 );
133 expect(resolved.traces?.enabled).toBe(false);
134 expect(resolved.traces?.persist).toBeUndefined();
135 });
136
137 unit("fills traces when news.observability exists without traces", () => {
138 const resolved = resolveObservability(
139 {
140 observability: {
141 enabled: true,
142 logs: { enabled: true, invocationLogs: true, persist: true },
143 },
144 },
145 tracesBind({ enabled: true, headSamplingRate: 1 }),
146 );
147 expect(resolved.logs?.persist).toBe(true);
148 expect(resolved.traces?.enabled).toBe(true);
149 expect(resolved.traces?.headSamplingRate).toBe(1);
150 });
151
152 unit("no bind returns default logs", () => {
153 const resolved = resolveObservability({}, []);
154 expect(resolved.traces).toBeUndefined();
155 expect(resolved.logs?.invocationLogs).toBe(true);
156 });
157 },
158);
159
160test.provider(
161 "Cloudflare.Telemetry() enables traces without clobbering default logs",
162 (stack) =>
163 Effect.gen(function* () {
164 const { accountId } = yield* yield* CloudflareEnvironment;
165 yield* stack.destroy();
166
167 const worker = yield* stack.deploy(
168 Effect.gen(function* () {
169 return yield* NativeTracingWorker;
170 }),
171 );
172
173 const settings = yield* workers.getScriptScriptAndVersionSetting({
174 accountId,
175 scriptName: worker.workerName,
176 });
177 expect(settings.observability?.enabled).toBe(true);
178 expect(settings.observability?.logs?.enabled).toBe(true);
179 expect(settings.observability?.logs?.invocationLogs).toBe(true);
180 expect(settings.observability?.traces?.enabled).toBe(true);
181
182 yield* expectUrlContains(`${worker.url}/work`, "native-did-work", {
183 timeout: "180 seconds",
184 });
185
186 yield* stack.destroy();
187 yield* waitForWorkerToBeDeleted(worker.workerName, accountId);
188 }).pipe(logLevel),
189 {
190 tags: [
191 "provider:cloudflare",
192 "provider:cloudflare:kv",
193 "provider:cloudflare:queue",
194 "provider:cloudflare:worker",
195 "live",
196 ],
197 timeout: 180_000,
198 },
199);
200
201test.provider(
202 "explicit observability.traces wins over Cloudflare.Telemetry() bind",
203 (stack) =>
204 Effect.gen(function* () {
205 const { accountId } = yield* yield* CloudflareEnvironment;
206 yield* stack.destroy();
207
208 const worker = yield* stack.deploy(
209 Effect.gen(function* () {
210 return yield* makeTracedWorker("NativeTracingOverride", {
211 observability: {
212 enabled: true,
213 traces: { enabled: false },
214 },
215 });
216 }),
217 );
218
219 const settings = yield* workers.getScriptScriptAndVersionSetting({
220 accountId,
221 scriptName: worker.workerName,
222 });
223 expect(settings.observability?.traces?.enabled).toBe(false);
224 expect(settings.observability?.logs?.invocationLogs).toBe(true);
225
226 yield* stack.destroy();
227 yield* waitForWorkerToBeDeleted(worker.workerName, accountId);
228 }).pipe(logLevel),
229 {
230 tags: [
231 "provider:cloudflare",
232 "provider:cloudflare:kv",
233 "provider:cloudflare:queue",
234 "provider:cloudflare:worker",
235 "live",
236 ],
237 timeout: 180_000,
238 },
239);
240
241test.provider(
242 "Cloudflare.Telemetry() fails deploy on a pre-startActiveSpan compatibility date",
243 (stack) =>
244 Effect.gen(function* () {
245 yield* stack.destroy();
246
247 const error = yield* stack
248 .deploy(
249 Effect.gen(function* () {
250 return yield* makeTracedWorker("NativeTracingOldDate", {
251 compatibility: { date: "2026-03-17" },
252 });
253 }),
254 )
255 .pipe(Effect.flip);
256
257 expect(error).toBeInstanceOf(CloudflareTelemetryCompatibilityError);
258 expect(String(error)).toContain(MIN_CLOUDFLARE_TRACING_DATE);
259 expect(String(error)).toContain("2026-03-17");
260
261 yield* stack.destroy();
262 }).pipe(logLevel),
263 {
264 tags: [
265 "provider:cloudflare",
266 "provider:cloudflare:kv",
267 "provider:cloudflare:queue",
268 "provider:cloudflare:worker",
269 "live",
270 ],
271 timeout: 120_000,
272 },
273);
274
275test.provider(
276 "Effect spans appear in Workers Observability after Cloudflare.Telemetry()",
277 (stack) =>
278 Effect.gen(function* () {
279 const { accountId } = yield* yield* CloudflareEnvironment;
280 yield* stack.destroy();
281
282 const worker = yield* stack.deploy(
283 Effect.gen(function* () {
284 return yield* NativeTracingWorker;
285 }),
286 );
287
288 yield* expectUrlContains(`${worker.url}/work`, "native-did-work", {
289 timeout: "180 seconds",
290 });
291
292 // Poll until the request's Effect spans have been ingested.
293 const spans = yield* querySpans(accountId, worker.workerName).pipe(
294 Effect.repeat({
295 schedule: Schedule.spaced("5 seconds"),
296 until: (spans) =>
297 spans.some((s) => s.spanName === "operation") &&
298 spans.some((s) => s.spanName === "native.child"),
299 times: 36,
300 }),
301 );
302 expect(spans.map((s) => s.spanName)).toEqual(
303 expect.arrayContaining(["operation", "native.child"]),
304 );
305
306 yield* stack.destroy();
307 yield* waitForWorkerToBeDeleted(worker.workerName, accountId);
308 }).pipe(logLevel),
309 {
310 tags: [
311 "provider:cloudflare",
312 "provider:cloudflare:kv",
313 "provider:cloudflare:queue",
314 "provider:cloudflare:worker",
315 "live",
316 ],
317 timeout: 300_000,
318 },
319);
320
321const FANOUT_SPANS = [
322 "operation",
323 "child.a",
324 "child.a.inner",
325 "child.b",
326 "child.b.inner",
327 "forked",
328 "forked.inner",
329];
330
331test.provider(
332 "records the Effect span tree across fibers, exits, Durable Object and queue handlers",
333 (stack) =>
334 Effect.gen(function* () {
335 const { accountId } = yield* yield* CloudflareEnvironment;
336 yield* stack.destroy();
337
338 const worker = yield* stack.deploy(
339 Effect.gen(function* () {
340 return yield* NativeTracingWorker;
341 }),
342 );
343 yield* expectUrlContains(
344 `${worker.url}/fanout?id=warmup`,
345 "native-did-fanout",
346 { timeout: "180 seconds" },
347 );
348
349 // Several fan-out invocations in flight at once so fibers from
350 // different requests interleave inside the isolate as well, plus one
351 // request per remaining event path. Each request retries through
352 // workers.dev propagation (a colo can still serve Cloudflare's error
353 // page right after the warm-up succeeded elsewhere); a retried id just
354 // produces one more independent trace.
355 const ids = yield* Effect.sync(() => ({
356 fanout: Array.from({ length: 3 }, () => globalThis.crypto.randomUUID()),
357 exits: globalThis.crypto.randomUUID(),
358 rpc: globalThis.crypto.randomUUID(),
359 queue: globalThis.crypto.randomUUID(),
360 }));
361 const hit = (path: string, id: string, marker = id) =>
362 expectUrlContains(`${worker.url}${path}?id=${id}`, marker, {
363 timeout: "120 seconds",
364 });
365 yield* Effect.all(
366 [
367 ...ids.fanout.map((id) => hit("/fanout", id)),
368 hit("/exits", ids.exits),
369 hit("/rpc", ids.rpc, "native-did-rpc:do-ok"),
370 hit("/enqueue", ids.queue),
371 // Head sampling 1: every Effect span is sampled.
372 expectUrlContains(
373 `${worker.url}/sampled`,
374 '"operation":true,"child":true',
375 { timeout: "120 seconds" },
376 ),
377 ],
378 { concurrency: "unbounded" },
379 );
380
381 // `request.id` is annotated on each root span, so a request's trace
382 // is located via `$metadata.requestId`; wait until every expected
383 // span of every request has been ingested.
384 const expected = [
385 ...ids.fanout.map((id) => ({
386 root: "operation",
387 id,
388 names: FANOUT_SPANS,
389 })),
390 {
391 root: "operation",
392 id: ids.exits,
393 names: ["operation", "failing.child", "interrupted.child"],
394 },
395 {
396 root: "do.operation",
397 id: ids.rpc,
398 names: ["do.operation", "do.inner"],
399 },
400 {
401 root: "queue.operation",
402 id: ids.queue,
403 names: ["queue.operation", "queue.inner"],
404 },
405 ];
406 const ingested = (spans: SpanRow[]) =>
407 expected.every((e) => {
408 const trace = traceOf(spans, e.root, e.id);
409 return (
410 trace !== undefined &&
411 e.names.every((name) =>
412 trace.spans.some((s) => s.spanName === name),
413 )
414 );
415 });
416 // Ingestion latency varies from seconds to several minutes.
417 const spans = yield* querySpans(accountId, worker.workerName).pipe(
418 Effect.repeat({
419 schedule: Schedule.spaced("5 seconds"),
420 until: ingested,
421 times: 60,
422 }),
423 );
424 if (!ingested(spans)) {
425 yield* Effect.logError(
426 `native-tracing spans ingested so far: ${JSON.stringify(
427 spans.map((s) => [s.requestId, s.spanName]),
428 )}`,
429 );
430 }
431 expect(ingested(spans)).toBe(true);
432
433 // Fan-out: siblings and nesting follow the Effect span tree, and each
434 // KV read (a Cloudflare auto-instrumented span) is attributed to the
435 // branch whose fiber issued it — even though the scheduler resumed
436 // that fiber on a timer tick outside `operation`'s async context.
437 for (const id of ids.fanout) {
438 const trace = traceOf(spans, "operation", id)!;
439 const operation = trace.root;
440 // `operation` hangs off Cloudflare's own request span.
441 expect(
442 trace.spans.some(
443 (s) =>
444 s.spanId === operation.parentSpanId && s.spanName !== "operation",
445 ),
446 ).toBe(true);
447 for (const name of ["child.a", "child.b", "forked"]) {
448 const branch = trace.span(name);
449 expect(branch.parentSpanId).toBe(operation.spanId);
450 expect(trace.span(`${name}.inner`).parentSpanId).toBe(branch.spanId);
451 const platform = trace.spans.filter(
452 (s) =>
453 s.parentSpanId === branch.spanId &&
454 !FANOUT_SPANS.includes(s.spanName ?? ""),
455 );
456 expect(platform.length).toBeGreaterThanOrEqual(1);
457 }
458 expect(operation.exit).toBe("success");
459 }
460
461 // Exits: the failing child (caught by its parent) and the interrupted
462 // child record their outcome as `effect.exit`.
463 {
464 const trace = traceOf(spans, "operation", ids.exits)!;
465 expect(trace.root.exit).toBe("success");
466 const failing = trace.span("failing.child");
467 expect(failing.parentSpanId).toBe(trace.root.spanId);
468 expect(failing.exit).toBe("failure");
469 const interrupted = trace.span("interrupted.child");
470 expect(interrupted.parentSpanId).toBe(trace.root.spanId);
471 expect(interrupted.exit).toBe("interrupted");
472 }
473
474 // Durable Object RPC: spans opened inside the DO method nest under
475 // the DO invocation's platform span.
476 {
477 const trace = traceOf(spans, "do.operation", ids.rpc)!;
478 expect(trace.root.parentSpanId).toBeTruthy();
479 expect(trace.span("do.inner").parentSpanId).toBe(trace.root.spanId);
480 expect(trace.root.exit).toBe("success");
481 }
482
483 // Queue consumer: same for the `queue` event handler.
484 {
485 const trace = traceOf(spans, "queue.operation", ids.queue)!;
486 expect(trace.root.parentSpanId).toBeTruthy();
487 expect(trace.span("queue.inner").parentSpanId).toBe(trace.root.spanId);
488 expect(trace.root.exit).toBe("success");
489 }
490
491 yield* stack.destroy();
492 yield* waitForWorkerToBeDeleted(worker.workerName, accountId);
493 }).pipe(logLevel),
494 {
495 tags: [
496 "provider:cloudflare",
497 "provider:cloudflare:kv",
498 "provider:cloudflare:queue",
499 "provider:cloudflare:worker",
500 "live",
501 ],
502 timeout: 480_000,
503 },
504);
505
506test.provider(
507 "head sampling below 1 is applied by Cloudflare at ingestion",
508 (stack) =>
509 Effect.gen(function* () {
510 const { accountId } = yield* yield* CloudflareEnvironment;
511 const currentConfig = yield* ConfigProvider.ConfigProvider;
512 yield* stack.destroy();
513
514 const worker = yield* stack
515 .deploy(
516 Effect.gen(function* () {
517 return yield* makeTracedWorker("NativeTracingSampled");
518 }),
519 )
520 .pipe(
521 Effect.provideService(
522 ConfigProvider.ConfigProvider,
523 ConfigProvider.orElse(
524 ConfigProvider.fromUnknown({ HEAD_SAMPLING_RATE: 0.5 }),
525 currentConfig,
526 ),
527 ),
528 );
529
530 const settings = yield* workers.getScriptScriptAndVersionSetting({
531 accountId,
532 scriptName: worker.workerName,
533 });
534 expect(settings.observability?.traces?.enabled).toBe(true);
535 expect(settings.observability?.traces?.headSamplingRate).toBe(0.5);
536
537 // The runtime reports every invocation as traced (`span.isTraced` does
538 // not carry the head-sampling decision — a rate of 0 behaves the same),
539 // so Effect's `sampled` flag stays true and consistent parent → child.
540 // Sampling is applied when Cloudflare ingests: of N invocations, only
541 // ~half persist.
542 //
543 // A 50% rate (rather than a few percent) keeps this test fast and
544 // two-sided: some request of the batch is always persisted, so the
545 // ingestion wait ends as soon as rows land instead of at its ceiling,
546 // and the count can be bounded from below as well as above.
547 const runId = yield* Effect.sync(() => globalThis.crypto.randomUUID());
548 yield* expectUrlContains(
549 `${worker.url}/sampled?id=warmup-${runId}`,
550 "native-did-sample",
551 { timeout: "180 seconds" },
552 );
553 const client = yield* HttpClient.HttpClient;
554 const probe = client.get(`${worker.url}/sampled?id=${runId}`).pipe(
555 Effect.flatMap((res) => res.json),
556 Effect.map((body) => body as { operation: boolean; child: boolean }),
557 Effect.retry({ schedule: Schedule.spaced("1 second"), times: 30 }),
558 );
559 const requests = 100;
560 const seen = yield* Effect.all(
561 Array.from({ length: requests }, () => probe),
562 { concurrency: 10 },
563 );
564 expect(seen.filter((b) => b.child !== b.operation)).toEqual([]);
565
566 // One `operation` root per persisted request, keyed by this run's id
567 // so earlier runs inside the 15-minute query window don't count.
568 // Rows land in batches, so wait until two consecutive reads agree
569 // (and enough exist to prove tracing is on) before bounding the count.
570 // Binomial(100, 0.5) is within [20, 80] with P(miss) ≈ 1e-9; a few
571 // retried probes add at most a handful of extra traces.
572 const persistedCount = querySpans(accountId, worker.workerName).pipe(
573 Effect.map(
574 (spans) =>
575 spans.filter(
576 (s) => s.spanName === "operation" && s.requestId === runId,
577 ).length,
578 ),
579 );
580 let previous = -1;
581 const persisted = yield* persistedCount.pipe(
582 Effect.repeat({
583 schedule: Schedule.spaced("5 seconds"),
584 until: (count) => {
585 const settled = count >= 20 && count === previous;
586 previous = count;
587 return settled;
588 },
589 times: 36,
590 }),
591 );
592 expect(persisted).toBeGreaterThanOrEqual(20);
593 expect(persisted).toBeLessThanOrEqual(80);
594
595 yield* stack.destroy();
596 yield* waitForWorkerToBeDeleted(worker.workerName, accountId);
597 }).pipe(logLevel),
598 {
599 tags: [
600 "provider:cloudflare",
601 "provider:cloudflare:kv",
602 "provider:cloudflare:queue",
603 "provider:cloudflare:worker",
604 "live",
605 ],
606 timeout: 300_000,
607 },
608);