import fs from "node:fs";import os from "node:os";import path from "node:path"; import { assert, describe, it } from "@effect/vitest";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Logger from "effect/Logger";import * as References from "effect/References";import * as Schema from "effect/Schema";import * as Tracer from "effect/Tracer"; import { compactTraceAttributes, makeLocalFileTracer, makeTraceSink, type TraceRecord,} from "./observability.ts"; const TraceRecordLine = Schema.Struct({ name: Schema.String, spanId: Schema.String, parentSpanId: Schema.optional(Schema.String), attributes: Schema.Record(Schema.String, Schema.Unknown), events: Schema.Array( Schema.Struct({ name: Schema.String, attributes: Schema.Record(Schema.String, Schema.Unknown), }), ), exit: Schema.optional( Schema.Struct({ _tag: Schema.String, }), ),}); const decodeTraceRecordLine = Schema.decodeUnknownSync(Schema.fromJsonString(TraceRecordLine)); const makeRecord = (name: string, suffix = ""): TraceRecord => ({ type: "effect-span", name, traceId: `trace-${name}-${suffix}`, spanId: `span-${name}-${suffix}`, sampled: true, kind: "internal", startTimeUnixNano: "1", endTimeUnixNano: "2", durationMs: 1, attributes: { payload: suffix, }, events: [], links: [], exit: { _tag: "Success", },}); const readTraceRecords = (tracePath: string) => fs .readFileSync(tracePath, "utf8") .trim() .split("\n") .filter((line) => line.length > 0) .map((line) => decodeTraceRecordLine(line)); const makeTestLayer = (tracePath: string) => Layer.mergeAll( Layer.effect( Tracer.Tracer, makeLocalFileTracer({ filePath: tracePath, maxBytes: 1024 * 1024, maxFiles: 2, batchWindowMs: 10_000, }), ), Logger.layer([Logger.tracerLogger], { mergeWithExisting: false }), Layer.succeed(References.MinimumLogLevel, "Info"), ); describe("observability", () => { it("normalizes circular arrays, maps, and sets without recursing forever", () => { const array: Array<unknown> = ["alpha"]; array.push(array); const map = new Map<string, unknown>(); map.set("self", map); const set = new Set<unknown>(); set.add(set); assert.deepStrictEqual( compactTraceAttributes({ array, map, set, }), { array: ["alpha", "[Circular]"], map: { self: "[Circular]" }, set: ["[Circular]"], }, ); }); it("normalizes invalid dates without throwing", () => { assert.deepStrictEqual( compactTraceAttributes({ invalidDate: new Date("not-a-real-date"), }), { invalidDate: "Invalid Date", }, ); }); it.effect("flushes buffered trace records on close", () => Effect.scoped( Effect.gen(function* () { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-trace-sink-")); const tracePath = path.join(tempDir, "shared.trace.ndjson"); try { const sink = yield* makeTraceSink({ filePath: tracePath, maxBytes: 1024, maxFiles: 2, batchWindowMs: 10_000, }); sink.push(makeRecord("alpha")); sink.push(makeRecord("beta")); yield* sink.close(); const lines = readTraceRecords(tracePath); assert.equal(lines.length, 2); assert.equal(lines[0]?.name, "alpha"); assert.equal(lines[1]?.name, "beta"); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); } }), ), ); it.effect("rotates the trace file when the configured max size is exceeded", () => Effect.scoped( Effect.gen(function* () { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-trace-sink-")); const tracePath = path.join(tempDir, "shared.trace.ndjson"); try { const sink = yield* makeTraceSink({ filePath: tracePath, maxBytes: 180, maxFiles: 2, batchWindowMs: 10_000, }); for (let index = 0; index < 8; index += 1) { sink.push(makeRecord("rotate", `${index}-${"x".repeat(48)}`)); yield* sink.flush; } yield* sink.close(); const matchingFiles = fs .readdirSync(tempDir) .filter( (entry) => entry === "shared.trace.ndjson" || entry.startsWith("shared.trace.ndjson."), ) .toSorted(); assert.equal( matchingFiles.some((entry) => entry === "shared.trace.ndjson.1"), true, ); assert.equal( matchingFiles.some((entry) => entry === "shared.trace.ndjson.3"), false, ); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); } }), ), ); it.effect("drops only the invalid trace record when serialization fails", () => Effect.scoped( Effect.gen(function* () { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-trace-sink-")); const tracePath = path.join(tempDir, "shared.trace.ndjson"); try { const sink = yield* makeTraceSink({ filePath: tracePath, maxBytes: 1024, maxFiles: 2, batchWindowMs: 10_000, }); const circular: Array<unknown> = []; circular.push(circular); sink.push(makeRecord("alpha")); sink.push({ ...makeRecord("invalid"), attributes: { circular, }, } as TraceRecord); sink.push(makeRecord("beta")); yield* sink.close(); const lines = readTraceRecords(tracePath); assert.deepStrictEqual( lines.map((line) => line.name), ["alpha", "beta"], ); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); } }), ), ); it.effect("writes nested spans to disk and captures log messages as span events", () => Effect.gen(function* () { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-local-tracer-")); const tracePath = path.join(tempDir, "shared.trace.ndjson"); try { yield* Effect.scoped( Effect.gen(function* () { const program = Effect.gen(function* () { yield* Effect.annotateCurrentSpan({ "demo.parent": true, }); yield* Effect.logInfo("parent event"); yield* Effect.gen(function* () { yield* Effect.annotateCurrentSpan({ "demo.child": true, }); yield* Effect.logInfo("child event"); }).pipe(Effect.withSpan("child-span")); }).pipe(Effect.withSpan("parent-span")); yield* program.pipe(Effect.provide(makeTestLayer(tracePath))); }), ); const records = readTraceRecords(tracePath); assert.equal(records.length, 2); const parent = records.find((record) => record.name === "parent-span"); const child = records.find((record) => record.name === "child-span"); assert.notEqual(parent, undefined); assert.notEqual(child, undefined); if (!parent || !child) { return; } assert.equal(child.parentSpanId, parent.spanId); assert.equal(parent.attributes["demo.parent"], true); assert.equal(child.attributes["demo.child"], true); assert.equal( parent.events.some((event) => event.name === "parent event"), true, ); assert.equal( child.events.some((event) => event.name === "child event"), true, ); assert.equal( child.events.some((event) => event.attributes["effect.logLevel"] === "INFO"), true, ); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); } }), ); it.effect("serializes interrupted spans with an interrupted exit status", () => Effect.gen(function* () { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-local-tracer-")); const tracePath = path.join(tempDir, "shared.trace.ndjson"); try { yield* Effect.scoped( Effect.exit( Effect.interrupt.pipe( Effect.withSpan("interrupt-span"), Effect.provide(makeTestLayer(tracePath)), ), ), ); const records = readTraceRecords(tracePath); assert.equal(records.length, 1); assert.equal(records[0]?.name, "interrupt-span"); assert.equal(records[0]?.exit?._tag, "Interrupted"); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); } }), );});
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { assert, describe, it } from "@effect/vitest";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Logger from "effect/Logger";
import * as References from "effect/References";
import * as Schema from "effect/Schema";
import * as Tracer from "effect/Tracer";
import {
compactTraceAttributes,
makeLocalFileTracer,
makeTraceSink,
type TraceRecord,
} from "./observability.ts";
const TraceRecordLine = Schema.Struct({
name: Schema.String,
spanId: Schema.String,
parentSpanId: Schema.optional(Schema.String),
attributes: Schema.Record(Schema.String, Schema.Unknown),
events: Schema.Array(
Schema.Struct({
name: Schema.String,
attributes: Schema.Record(Schema.String, Schema.Unknown),
}),
),
exit: Schema.optional(
Schema.Struct({
_tag: Schema.String,
}),
),
});
const decodeTraceRecordLine = Schema.decodeUnknownSync(Schema.fromJsonString(TraceRecordLine));
const makeRecord = (name: string, suffix = ""): TraceRecord => ({
type: "effect-span",
name,
traceId: `trace-${name}-${suffix}`,
spanId: `span-${name}-${suffix}`,
sampled: true,
kind: "internal",
startTimeUnixNano: "1",
endTimeUnixNano: "2",
durationMs: 1,
attributes: {
payload: suffix,
},
events: [],
links: [],
exit: {
_tag: "Success",
},
});
const readTraceRecords = (tracePath: string) =>
fs
.readFileSync(tracePath, "utf8")
.trim()
.split("\n")
.filter((line) => line.length > 0)
.map((line) => decodeTraceRecordLine(line));
const makeTestLayer = (tracePath: string) =>
Layer.mergeAll(
Layer.effect(
Tracer.Tracer,
makeLocalFileTracer({
filePath: tracePath,
maxBytes: 1024 * 1024,
maxFiles: 2,
batchWindowMs: 10_000,
}),
),
Logger.layer([Logger.tracerLogger], { mergeWithExisting: false }),
Layer.succeed(References.MinimumLogLevel, "Info"),
);
describe("observability", () => {
it("normalizes circular arrays, maps, and sets without recursing forever", () => {
const array: Array<unknown> = ["alpha"];
array.push(array);
const map = new Map<string, unknown>();
map.set("self", map);
const set = new Set<unknown>();
set.add(set);
assert.deepStrictEqual(
compactTraceAttributes({
array,
map,
set,
}),
{
array: ["alpha", "[Circular]"],
map: { self: "[Circular]" },
set: ["[Circular]"],
},
);
});
it("normalizes invalid dates without throwing", () => {
assert.deepStrictEqual(
compactTraceAttributes({
invalidDate: new Date("not-a-real-date"),
}),
{
invalidDate: "Invalid Date",
},
);
});
it.effect("flushes buffered trace records on close", () =>
Effect.scoped(
Effect.gen(function* () {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-trace-sink-"));
const tracePath = path.join(tempDir, "shared.trace.ndjson");
try {
const sink = yield* makeTraceSink({
filePath: tracePath,
maxBytes: 1024,
maxFiles: 2,
batchWindowMs: 10_000,
});
sink.push(makeRecord("alpha"));
sink.push(makeRecord("beta"));
yield* sink.close();
const lines = readTraceRecords(tracePath);
assert.equal(lines.length, 2);
assert.equal(lines[0]?.name, "alpha");
assert.equal(lines[1]?.name, "beta");
} finally {
fs.rmSync(tempDir, { recursive: true, force: true });
}
}),
),
);
it.effect("rotates the trace file when the configured max size is exceeded", () =>
Effect.scoped(
Effect.gen(function* () {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-trace-sink-"));
const tracePath = path.join(tempDir, "shared.trace.ndjson");
try {
const sink = yield* makeTraceSink({
filePath: tracePath,
maxBytes: 180,
maxFiles: 2,
batchWindowMs: 10_000,
});
for (let index = 0; index < 8; index += 1) {
sink.push(makeRecord("rotate", `${index}-${"x".repeat(48)}`));
yield* sink.flush;
}
yield* sink.close();
const matchingFiles = fs
.readdirSync(tempDir)
.filter(
(entry) =>
entry === "shared.trace.ndjson" || entry.startsWith("shared.trace.ndjson."),
)
.toSorted();
assert.equal(
matchingFiles.some((entry) => entry === "shared.trace.ndjson.1"),
true,
);
assert.equal(
matchingFiles.some((entry) => entry === "shared.trace.ndjson.3"),
false,
);
} finally {
fs.rmSync(tempDir, { recursive: true, force: true });
}
}),
),
);
it.effect("drops only the invalid trace record when serialization fails", () =>
Effect.scoped(
Effect.gen(function* () {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-trace-sink-"));
const tracePath = path.join(tempDir, "shared.trace.ndjson");
try {
const sink = yield* makeTraceSink({
filePath: tracePath,
maxBytes: 1024,
maxFiles: 2,
batchWindowMs: 10_000,
});
const circular: Array<unknown> = [];
circular.push(circular);
sink.push(makeRecord("alpha"));
sink.push({
...makeRecord("invalid"),
attributes: {
circular,
},
} as TraceRecord);
sink.push(makeRecord("beta"));
yield* sink.close();
const lines = readTraceRecords(tracePath);
assert.deepStrictEqual(
lines.map((line) => line.name),
["alpha", "beta"],
);
} finally {
fs.rmSync(tempDir, { recursive: true, force: true });
}
}),
),
);
it.effect("writes nested spans to disk and captures log messages as span events", () =>
Effect.gen(function* () {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-local-tracer-"));
const tracePath = path.join(tempDir, "shared.trace.ndjson");
try {
yield* Effect.scoped(
Effect.gen(function* () {
const program = Effect.gen(function* () {
yield* Effect.annotateCurrentSpan({
"demo.parent": true,
});
yield* Effect.logInfo("parent event");
yield* Effect.gen(function* () {
yield* Effect.annotateCurrentSpan({
"demo.child": true,
});
yield* Effect.logInfo("child event");
}).pipe(Effect.withSpan("child-span"));
}).pipe(Effect.withSpan("parent-span"));
yield* program.pipe(Effect.provide(makeTestLayer(tracePath)));
}),
);
const records = readTraceRecords(tracePath);
assert.equal(records.length, 2);
const parent = records.find((record) => record.name === "parent-span");
const child = records.find((record) => record.name === "child-span");
assert.notEqual(parent, undefined);
assert.notEqual(child, undefined);
if (!parent || !child) {
return;
}
assert.equal(child.parentSpanId, parent.spanId);
assert.equal(parent.attributes["demo.parent"], true);
assert.equal(child.attributes["demo.child"], true);
assert.equal(
parent.events.some((event) => event.name === "parent event"),
true,
);
assert.equal(
child.events.some((event) => event.name === "child event"),
true,
);
assert.equal(
child.events.some((event) => event.attributes["effect.logLevel"] === "INFO"),
true,
);
} finally {
fs.rmSync(tempDir, { recursive: true, force: true });
}
}),
);
it.effect("serializes interrupted spans with an interrupted exit status", () =>
Effect.gen(function* () {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-local-tracer-"));
const tracePath = path.join(tempDir, "shared.trace.ndjson");
try {
yield* Effect.scoped(
Effect.exit(
Effect.interrupt.pipe(
Effect.withSpan("interrupt-span"),
Effect.provide(makeTestLayer(tracePath)),
),
),
);
const records = readTraceRecords(tracePath);
assert.equal(records.length, 1);
assert.equal(records[0]?.name, "interrupt-span");
assert.equal(records[0]?.exit?._tag, "Interrupted");
} finally {
fs.rmSync(tempDir, { recursive: true, force: true });
}
}),
);
});