Open History to see the last commit
1import { assert, describe, expect, it } from "@effect/vitest";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import * as Cause from "effect/Cause";
4import * as Effect from "effect/Effect";
5import * as Deferred from "effect/Deferred";
6import * as Fiber from "effect/Fiber";
7import * as Exit from "effect/Exit";
8import * as FileSystem from "effect/FileSystem";
9import { cast } from "effect/Function";
10import * as Layer from "effect/Layer";
11import * as Logger from "effect/Logger";
12import * as Path from "effect/Path";
13import * as Ref from "effect/Ref";
14import * as References from "effect/References";
15import * as Schema from "effect/Schema";
16import * as Tracer from "effect/Tracer";
17import type { OtlpTracer } from "effect/observability";
18
19import {
20 causeErrorTag,
21 compactTraceAttributes,
22 decodeOtlpTraceRecords,
23 errorTag,
24 makeLocalFileTracer,
25 makeTraceSink,
26 type EffectTraceRecord,
27 type TraceRecord,
28 OtlpHeadersFromString,
29 truncateTraceAttributes,
30 processServiceInstanceId,
31} from "./observability.ts";
32
33describe("processServiceInstanceId", () => {
34 it("stays the same for the process unless the operator set one", () => {
35 expect(processServiceInstanceId()).toBe(processServiceInstanceId({}));
36 expect(processServiceInstanceId({ "service.instance.id": "pod-7" })).toBe("pod-7");
37 });
38});
39
40describe("errorTag", () => {
41 it("reports structural tags without retaining arbitrary values", () => {
42 assert.equal(errorTag({ _tag: "AcpRequestError" }), "AcpRequestError");
43 assert.equal(errorTag(new TypeError("secret-token-value")), "TypeError");
44 assert.equal(errorTag({ _tag: "secret token value" }), "TaggedError");
45 });
46});
47
48describe("causeErrorTag", () => {
49 it("reports the tagged failure value instead of the Cause reason wrapper", () => {
50 assert.equal(
51 causeErrorTag(Cause.fail({ _tag: "ServerAuthInvalidCredentialError" })),
52 "ServerAuthInvalidCredentialError",
53 );
54 });
55});
56
57const TraceRecordLine = Schema.Struct({
58 name: Schema.String,
59 spanId: Schema.String,
60 parentSpanId: Schema.optional(Schema.String),
61 attributes: Schema.Record(Schema.String, Schema.Unknown),
62 events: Schema.Array(
63 Schema.Struct({
64 name: Schema.String,
65 attributes: Schema.Record(Schema.String, Schema.Unknown),
66 }),
67 ),
68 exit: Schema.optional(
69 Schema.Struct({
70 _tag: Schema.String,
71 }),
72 ),
73});
74
75const decodeTraceRecordLine = Schema.decodeUnknownSync(Schema.fromJsonString(TraceRecordLine));
76
77const makeRecord = (name: string, suffix = ""): TraceRecord => ({
78 type: "effect-span",
79 name,
80 traceId: `trace-${name}-${suffix}`,
81 spanId: `span-${name}-${suffix}`,
82 sampled: true,
83 kind: "internal",
84 startTimeUnixNano: "1",
85 endTimeUnixNano: "2",
86 durationMs: 1,
87 attributes: {
88 payload: suffix,
89 },
90 events: [],
91 links: [],
92 exit: {
93 _tag: "Success",
94 },
95});
96
97const readTraceRecords = Effect.fn("readTraceRecords")(function* (tracePath: string) {
98 const fileSystem = yield* FileSystem.FileSystem;
99 return (yield* fileSystem.readFileString(tracePath))
100 .trim()
101 .split("\n")
102 .filter((line) => line.length > 0)
103 .map((line) => decodeTraceRecordLine(line));
104});
105
106const layerTest = (tracePath: string) =>
107 Layer.mergeAll(
108 Layer.effect(
109 Tracer.Tracer,
110 makeLocalFileTracer({
111 filePath: tracePath,
112 maxBytes: 1024 * 1024,
113 maxFiles: 2,
114 batchWindowMs: 10_000,
115 }),
116 ),
117 Logger.layer([Logger.tracerLogger], { mergeWithExisting: false }),
118 Layer.succeed(References.MinimumLogLevel, "Info"),
119 );
120
121const nodeServicesIt = it.layer(NodeServices.layer);
122
123describe("truncateTraceAttributes", () => {
124 it("clamps oversized strings at any depth without mutating the input", () => {
125 const stack = "s".repeat(2_000);
126 const attributes = {
127 "db.query.text": "q".repeat(2_000),
128 short: "ok",
129 error: { name: "Error", stack, nested: ["a".repeat(2_000)] },
130 };
131 const truncated = truncateTraceAttributes(attributes);
132
133 assert.equal((truncated["db.query.text"] as string).length, 200 + "…[truncated]".length);
134 assert.equal(truncated["short"], "ok");
135 const error = truncated["error"] as { stack: string; nested: Array<string> };
136 assert.equal(error.stack.length, 500 + "…[truncated]".length);
137 assert.equal(error.nested[0]?.length, 500 + "…[truncated]".length);
138 assert.equal(attributes.error.stack, stack);
139 });
140});
141
142describe("observability", () => {
143 it.effect(
144 "preserves successful results without retaining them in sampled or unsampled spans",
145 () =>
146 Effect.gen(function* () {
147 for (const sampled of [true, false]) {
148 const delegates: Array<Tracer.Span> = [];
149 const records: Array<EffectTraceRecord> = [];
150 const tracer = yield* makeLocalFileTracer({
151 filePath: "unused",
152 maxBytes: 1024,
153 maxFiles: 1,
154 batchWindowMs: 10_000,
155 sink: {
156 filePath: "unused",
157 push: (record) => {
158 if (record.type === "effect-span") records.push(record);
159 },
160 flush: Effect.void,
161 flushSync: () => {},
162 close: () => Effect.void,
163 },
164 delegate: Tracer.make({
165 span: (options) => {
166 const span = new Tracer.NativeSpan({ ...options, sampled });
167 delegates.push(span);
168 return span;
169 },
170 }),
171 });
172 const payload = { turnItems: [{ output: [{ text: "synthetic-result" }] }] };
173 let span: Tracer.Span | undefined;
174 const result = yield* Effect.gen(function* () {
175 span = yield* Effect.currentSpan;
176 return payload;
177 }).pipe(
178 Effect.withSpan("read-thread-projection"),
179 Effect.provideService(Tracer.Tracer, tracer),
180 );
181 assert.strictEqual(result, payload);
182 assert.isDefined(span);
183 for (const completed of [span!, ...delegates]) {
184 assert.equal(completed.status._tag, "Ended");
185 if (completed.status._tag === "Ended") {
186 assert.deepStrictEqual(completed.status.exit, Exit.void);
187 }
188 }
189 assert.deepStrictEqual(
190 records.map((record) => record.exit),
191 sampled ? [{ _tag: "Success" }] : [],
192 );
193 }
194 }).pipe(Effect.provide(NodeServices.layer)),
195 );
196
197 it("normalizes circular arrays, maps, and sets without recursing forever", () => {
198 const array: Array<unknown> = ["alpha"];
199 array.push(array);
200
201 const map = new Map<string, unknown>();
202 map.set("self", map);
203
204 const set = new Set<unknown>();
205 set.add(set);
206
207 assert.deepStrictEqual(
208 compactTraceAttributes({
209 array,
210 map,
211 set,
212 }),
213 {
214 array: ["alpha", "[Circular]"],
215 map: { self: "[Circular]" },
216 set: ["[Circular]"],
217 },
218 );
219 });
220
221 it("keeps an object that appears twice without marking it circular", () => {
222 const shared = { cwd: "/repo" };
223 const sharedMap = new Map([["cwd", "/repo"]]);
224 const sharedSet = new Set(["/repo"]);
225
226 assert.deepStrictEqual(
227 compactTraceAttributes({
228 ctx: { first: shared, second: shared },
229 list: [shared, shared],
230 maps: [sharedMap, sharedMap],
231 sets: [sharedSet, sharedSet],
232 }),
233 {
234 ctx: { first: { cwd: "/repo" }, second: { cwd: "/repo" } },
235 list: [{ cwd: "/repo" }, { cwd: "/repo" }],
236 maps: [{ cwd: "/repo" }, { cwd: "/repo" }],
237 sets: [["/repo"], ["/repo"]],
238 },
239 );
240 });
241
242 it("caps shared references that would expand exponentially", () => {
243 let graph: Record<string, unknown> = { cwd: "/repo" };
244 for (let depth = 0; depth < 20; depth++) {
245 graph = { left: graph, right: graph };
246 }
247
248 const serialized = JSON.stringify(compactTraceAttributes({ graph }));
249
250 assert.include(serialized, '"[Truncated]"');
251 assert.isBelow(serialized.length, 1_000_000);
252 });
253
254 nodeServicesIt("node services", (it) => {
255 it.effect(
256 "keeps trace producers responsive during a slow write, bounds the backlog, and drains on close",
257 () =>
258 Effect.gen(function* () {
259 const fs = yield* FileSystem.FileSystem;
260 const path = yield* Path.Path;
261 const directory = yield* fs.makeTempDirectoryScoped({
262 prefix: "t3-trace-backpressure-",
263 });
264 const tracePath = path.join(directory, "trace.ndjson");
265 const started = yield* Deferred.make<void>();
266 const release = yield* Deferred.make<void>();
267 const closed = yield* Ref.make(false);
268 const sink = yield* makeTraceSink({
269 filePath: tracePath,
270 maxBytes: 5_000_000,
271 maxFiles: 2,
272 batchWindowMs: 10_000,
273 }).pipe(
274 Effect.provideService(FileSystem.FileSystem, {
275 ...fs,
276 writeFile: (file, data, options) =>
277 Effect.gen(function* () {
278 yield* Deferred.succeed(started, undefined);
279 yield* Deferred.await(release);
280 yield* fs.writeFile(file, data, options);
281 }),
282 }),
283 );
284 sink.push(makeRecord("first"));
285 const flushing = yield* Effect.forkChild(sink.flush);
286 yield* Deferred.await(started);
287 for (let index = 0; index < 2_000; index++) sink.push(makeRecord("small", String(index)));
288 for (let index = 0; index < 3; index++)
289 sink.push({ ...makeRecord("large"), attributes: { payload: "x".repeat(2_000_000) } });
290 const closing = yield* Effect.forkChild(
291 sink.close().pipe(Effect.tap(() => Ref.set(closed, true))),
292 );
293 yield* Effect.yieldNow;
294 assert.isFalse(yield* Ref.get(closed));
295 yield* Deferred.succeed(release, undefined);
296 yield* Fiber.join(flushing);
297 yield* Fiber.join(closing);
298 const records = yield* readTraceRecords(tracePath);
299 assert.deepStrictEqual(
300 records.map((record) => record.name),
301 ["first", ...Array.from({ length: 2_000 }, () => "small"), "large", "large"],
302 );
303 assert.isTrue(yield* Ref.get(closed));
304 }),
305 );
306
307 it.effect("writes buffered records synchronously for process exit", () =>
308 Effect.gen(function* () {
309 const fileSystem = yield* FileSystem.FileSystem;
310 const path = yield* Path.Path;
311 const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-trace-sink-" });
312 const tracePath = path.join(tempDir, "exit.trace.ndjson");
313 const sink = yield* makeTraceSink({
314 filePath: tracePath,
315 maxBytes: 1_000_000,
316 maxFiles: 2,
317 batchWindowMs: 10_000,
318 });
319
320 sink.push(makeRecord("before-exit"));
321 sink.flushSync();
322 sink.flushSync();
323
324 const records = yield* readTraceRecords(tracePath);
325 assert.deepStrictEqual(
326 records.map((record) => record.name),
327 ["before-exit"],
328 );
329 }),
330 );
331
332 it.effect("keeps every trace file within the configured limit for threshold flushes", () =>
333 Effect.gen(function* () {
334 const fileSystem = yield* FileSystem.FileSystem;
335 const path = yield* Path.Path;
336 const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-trace-sink-" });
337 const tracePath = path.join(tempDir, "shared.trace.ndjson");
338 const maxBytes = 1_024;
339
340 const sink = yield* makeTraceSink({
341 filePath: tracePath,
342 maxBytes,
343 maxFiles: 2,
344 batchWindowMs: 10_000,
345 });
346
347 for (let index = 0; index < 256; index += 1) {
348 sink.push(makeRecord("threshold", `${index}-${"x".repeat(48)}`));
349 }
350 yield* sink.close();
351
352 const matchingFiles = (yield* fileSystem.readDirectory(tempDir)).filter(
353 (entry) => entry === "shared.trace.ndjson" || entry.startsWith("shared.trace.ndjson."),
354 );
355 assert.include(matchingFiles, "shared.trace.ndjson.1");
356 for (const entry of matchingFiles) {
357 const stat = yield* fileSystem.stat(path.join(tempDir, entry));
358 assert.isAtMost(Number(stat.size), maxBytes, entry);
359 }
360 }),
361 );
362
363 it.effect("drops a single trace record that cannot fit within the configured limit", () =>
364 Effect.gen(function* () {
365 const fileSystem = yield* FileSystem.FileSystem;
366 const path = yield* Path.Path;
367 const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-trace-sink-" });
368 const tracePath = path.join(tempDir, "shared.trace.ndjson");
369 const maxBytes = 1_024;
370
371 const sink = yield* makeTraceSink({
372 filePath: tracePath,
373 maxBytes,
374 maxFiles: 2,
375 batchWindowMs: 10_000,
376 });
377
378 sink.push(makeRecord("oversized", "x".repeat(maxBytes * 2)));
379 sink.push(makeRecord("retained"));
380 yield* sink.close();
381
382 const records = yield* readTraceRecords(tracePath);
383 const stat = yield* fileSystem.stat(tracePath);
384 assert.deepEqual(
385 records.map((record) => record.name),
386 ["retained"],
387 );
388 assert.isAtMost(Number(stat.size), maxBytes);
389 }),
390 );
391
392 it.effect("keeps WebSocket credentials out of trace files and exported spans", () =>
393 Effect.gen(function* () {
394 const fileSystem = yield* FileSystem.FileSystem;
395 const path = yield* Path.Path;
396 const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-trace-sink-" });
397 const tracePath = path.join(tempDir, "shared.trace.ndjson");
398 const sink = yield* makeTraceSink({
399 filePath: tracePath,
400 maxBytes: 1024 * 1024,
401 maxFiles: 2,
402 batchWindowMs: 10_000,
403 });
404 const delegates: Array<Tracer.NativeSpan> = [];
405 const tracer = yield* makeLocalFileTracer({
406 filePath: tracePath,
407 maxBytes: 1024 * 1024,
408 maxFiles: 2,
409 batchWindowMs: 10_000,
410 sink,
411 delegate: Tracer.make({
412 span: (options) => {
413 const span = new Tracer.NativeSpan(options);
414 delegates.push(span);
415 return span;
416 },
417 }),
418 });
419 const url = "/ws?wsTicket=ticket-secret&dpop=proof-secret&clientSurface=web";
420
421 yield* Effect.annotateCurrentSpan({
422 "url.full": `http://127.0.0.1:3773${url}`,
423 "url.query": url.slice(4),
424 }).pipe(
425 Effect.withSpan("GET", { kind: "server" }),
426 Effect.provideService(Tracer.Tracer, tracer),
427 );
428 yield* Effect.die(new Error(`RequestParseError (GET ${url})`)).pipe(
429 Effect.tapCause((cause) => Effect.logError(cause)),
430 Effect.withSpan("DeviceHubProxy.proxyWebSocket"),
431 Effect.provideService(Tracer.Tracer, tracer),
432 Effect.provide(Logger.layer([Logger.tracerLogger], { mergeWithExisting: false })),
433 Effect.exit,
434 );
435 yield* sink.close();
436
437 const written = yield* fileSystem.readFileString(tracePath);
438 const exported = delegates.map((span) =>
439 JSON.stringify({
440 attributes: Object.fromEntries(span.attributes),
441 events: span.events.map(([, , attributes]) => attributes),
442 exit:
443 span.status._tag === "Ended" && Exit.isFailure(span.status.exit)
444 ? Cause.pretty(span.status.exit.cause)
445 : "",
446 }),
447 );
448 for (const text of [written, ...exported]) {
449 assert.notInclude(text, "ticket-secret");
450 assert.notInclude(text, "proof-secret");
451 }
452 assert.include(written, "wsTicket=<redacted>&dpop=<redacted>&clientSurface=web");
453 assert.include(exported.join("\n"), "effect.cause");
454 }),
455 );
456
457 it.effect("drops only the invalid trace record when serialization fails", () =>
458 Effect.gen(function* () {
459 const fileSystem = yield* FileSystem.FileSystem;
460 const path = yield* Path.Path;
461 const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-trace-sink-" });
462 const tracePath = path.join(tempDir, "shared.trace.ndjson");
463
464 const sink = yield* makeTraceSink({
465 filePath: tracePath,
466 maxBytes: 1024,
467 maxFiles: 2,
468 batchWindowMs: 10_000,
469 });
470
471 const circular: Array<unknown> = [];
472 circular.push(circular);
473
474 sink.push(makeRecord("alpha"));
475 sink.push({
476 ...makeRecord("invalid"),
477 attributes: {
478 circular,
479 },
480 } as TraceRecord);
481 sink.push(makeRecord("beta"));
482 yield* sink.close();
483
484 const lines = yield* readTraceRecords(tracePath);
485
486 assert.deepStrictEqual(
487 lines.map((line) => line.name),
488 ["alpha", "beta"],
489 );
490 }),
491 );
492
493 it.effect("writes nested spans to disk and captures log messages as span events", () =>
494 Effect.gen(function* () {
495 const fileSystem = yield* FileSystem.FileSystem;
496 const path = yield* Path.Path;
497 const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-local-tracer-" });
498 const tracePath = path.join(tempDir, "shared.trace.ndjson");
499
500 yield* Effect.scoped(
501 Effect.gen(function* () {
502 const program = Effect.gen(function* () {
503 yield* Effect.annotateCurrentSpan({
504 "demo.parent": true,
505 });
506 yield* Effect.logInfo("parent event");
507 yield* Effect.gen(function* () {
508 yield* Effect.annotateCurrentSpan({
509 "demo.child": true,
510 });
511 yield* Effect.logInfo("child event");
512 }).pipe(Effect.withSpan("child-span"));
513 }).pipe(Effect.withSpan("parent-span"));
514
515 yield* program.pipe(Effect.provide(layerTest(tracePath)));
516 }),
517 );
518
519 const records = yield* readTraceRecords(tracePath);
520 assert.equal(records.length, 2);
521
522 const parent = records.find((record) => record.name === "parent-span");
523 const child = records.find((record) => record.name === "child-span");
524
525 assert.notEqual(parent, undefined);
526 assert.notEqual(child, undefined);
527 if (!parent || !child) {
528 return;
529 }
530
531 assert.equal(child.parentSpanId, parent.spanId);
532 assert.equal(parent.attributes["demo.parent"], true);
533 assert.equal(child.attributes["demo.child"], true);
534 assert.equal(
535 parent.events.some((event) => event.name === "parent event"),
536 true,
537 );
538 assert.equal(
539 child.events.some((event) => event.name === "child event"),
540 true,
541 );
542 assert.equal(
543 child.events.some((event) => event.attributes["effect.logLevel"] === "INFO"),
544 true,
545 );
546 }),
547 );
548
549 it.effect("serializes interrupted spans with an interrupted exit status", () =>
550 Effect.gen(function* () {
551 const fileSystem = yield* FileSystem.FileSystem;
552 const path = yield* Path.Path;
553 const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-local-tracer-" });
554 const tracePath = path.join(tempDir, "shared.trace.ndjson");
555
556 yield* Effect.scoped(
557 Effect.exit(
558 Effect.interrupt.pipe(
559 Effect.withSpan("interrupt-span"),
560 Effect.provide(layerTest(tracePath)),
561 ),
562 ),
563 );
564
565 const records = yield* readTraceRecords(tracePath);
566 assert.equal(records.length, 1);
567 assert.equal(records[0]?.name, "interrupt-span");
568 assert.equal(records[0]?.exit?._tag, "Interrupted");
569 }),
570 );
571 });
572});
573
574describe("decodeOtlpTraceRecords", () => {
575 it("decodes spans when OTLP/JSON omits unset and empty fields", () => {
576 // Protobuf JSON leaves out unset and empty fields, so this batch is valid.
577 const payload = cast<unknown, OtlpTracer.TraceData>({
578 resourceSpans: [
579 {},
580 {
581 scopeSpans: [
582 {},
583 {
584 spans: [
585 {
586 traceId: "trace-1",
587 spanId: "span-1",
588 name: "browser.request",
589 startTimeUnixNano: "1000000",
590 endTimeUnixNano: "3000000",
591 attributes: [{ key: "tags", value: { arrayValue: {} } }],
592 },
593 ],
594 },
595 ],
596 },
597 ],
598 });
599
600 assert.deepStrictEqual(decodeOtlpTraceRecords(cast<unknown, OtlpTracer.TraceData>({})), []);
601 assert.deepStrictEqual(decodeOtlpTraceRecords(payload), [
602 {
603 type: "otlp-span",
604 name: "browser.request",
605 traceId: "trace-1",
606 spanId: "span-1",
607 sampled: true,
608 kind: "internal",
609 startTimeUnixNano: "1000000",
610 endTimeUnixNano: "3000000",
611 durationMs: 2,
612 attributes: { tags: [] },
613 resourceAttributes: {},
614 scope: { attributes: {} },
615 events: [],
616 links: [],
617 status: { code: "0" },
618 },
619 ]);
620 });
621});
622
623describe("OtlpHeadersFromString", () => {
624 const decode = Schema.decodeUnknownSync(OtlpHeadersFromString);
625
626 it.each([
627 {
628 name: "decodes percent-encoded values",
629 input: "authorization=Basic%20abc%3D%3D,x-tenant=t3",
630 expected: { authorization: "Basic abc==", "x-tenant": "t3" },
631 },
632 {
633 name: "ignores whitespace around separators",
634 input: "authorization=Basic%20abc%3D%3D, x-tenant = t3 ,",
635 expected: { authorization: "Basic abc==", "x-tenant": "t3" },
636 },
637 {
638 name: "keeps literal equals signs inside a value",
639 input: "authorization=Bearer abc==",
640 expected: { authorization: "Bearer abc==" },
641 },
642 {
643 name: "keeps an empty value",
644 input: "x-empty=",
645 expected: { "x-empty": "" },
646 },
647 ])("$name", ({ input, expected }) => {
648 expect(decode(input)).toEqual(expected);
649 });
650
651 it.each([
652 { name: "rejects a pair without a separator", input: "authorization" },
653 { name: "rejects a pair without a key", input: "=value" },
654 { name: "rejects a malformed percent-encoding", input: "authorization=%E0" },
655 ])("$name", ({ input }) => {
656 expect(() => decode(input)).toThrow();
657 });
658});