import * as Cause from "effect/Cause";import * as Effect from "effect/Effect";import type * as Exit from "effect/Exit";import * as ExitRuntime from "effect/Exit";import * as Option from "effect/Option";import * as Tracer from "effect/Tracer";import { OtlpResource, OtlpTracer } from "effect/unstable/observability"; import { RotatingFileSink } from "./logging.ts"; const FLUSH_BUFFER_THRESHOLD = 256;const textEncoder = new TextEncoder(); export type TraceAttributes = Readonly<Record<string, unknown>>; export interface TraceRecordEvent { readonly name: string; readonly timeUnixNano: string; readonly attributes: Readonly<Record<string, unknown>>;} export interface TraceRecordLink { readonly traceId: string; readonly spanId: string; readonly attributes: Readonly<Record<string, unknown>>;} interface BaseTraceRecord { readonly name: string; readonly kind: string; readonly traceId: string; readonly spanId: string; readonly parentSpanId?: string; readonly sampled: boolean; readonly startTimeUnixNano: string; readonly endTimeUnixNano: string; readonly durationMs: number; readonly attributes: Readonly<Record<string, unknown>>; readonly events: ReadonlyArray<TraceRecordEvent>; readonly links: ReadonlyArray<TraceRecordLink>;} export interface EffectTraceRecord extends BaseTraceRecord { readonly type: "effect-span"; readonly exit: | { readonly _tag: "Success"; } | { readonly _tag: "Interrupted"; readonly cause: string; } | { readonly _tag: "Failure"; readonly cause: string; };} export interface OtlpTraceRecord extends BaseTraceRecord { readonly type: "otlp-span"; readonly resourceAttributes: Readonly<Record<string, unknown>>; readonly scope: Readonly<{ readonly name?: string; readonly version?: string; readonly attributes: Readonly<Record<string, unknown>>; }>; readonly status?: | { readonly code?: string; readonly message?: string; } | undefined;} export type TraceRecord = EffectTraceRecord | OtlpTraceRecord; function isStructuralTag(value: unknown): value is string { return ( typeof value === "string" && value.length > 0 && value.length <= 128 && /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value) );} export function errorTag(error: unknown): string { try { if (typeof error === "object" && error !== null && "_tag" in error) { return isStructuralTag(error._tag) ? error._tag : "TaggedError"; } if (error instanceof Error) { return isStructuralTag(error.name) ? error.name : "Error"; } } catch { return "UnknownError"; } return typeof error;} export function causeErrorTag(cause: Cause.Cause<unknown>): string { const failure = Cause.findErrorOption(cause); if (Option.isSome(failure)) { return errorTag(failure.value); } return cause.reasons[0]?._tag ?? "Empty";} export interface TraceSinkOptions { readonly filePath: string; readonly maxBytes: number; readonly maxFiles: number; readonly batchWindowMs: number; readonly onFlush?: (stats: TraceSinkFlushStats) => Effect.Effect<void>;} export interface TraceSinkFlushStats { readonly logicalWriteBytes: number; readonly count: number; readonly durationMs: number;} export interface TraceSink { readonly filePath: string; push: (record: TraceRecord) => void; flush: Effect.Effect<void>; close: () => Effect.Effect<void>;} export interface LocalFileTracerOptions extends TraceSinkOptions { readonly delegate?: Tracer.Tracer; readonly sink?: TraceSink;} type OtlpSpan = OtlpTracer.ScopeSpan["spans"][number];type OtlpSpanEvent = OtlpSpan["events"][number];type OtlpSpanLink = OtlpSpan["links"][number];type OtlpSpanStatus = OtlpSpan["status"]; interface SerializableSpan { readonly name: string; readonly traceId: string; readonly spanId: string; readonly parent: Option.Option<Tracer.AnySpan>; readonly status: Tracer.SpanStatus; readonly sampled: boolean; readonly kind: Tracer.SpanKind; readonly attributes: ReadonlyMap<string, unknown>; readonly links: ReadonlyArray<Tracer.SpanLink>; readonly events: ReadonlyArray< readonly [name: string, startTime: bigint, attributes: Record<string, unknown>] >;} function isPlainObject(value: unknown): value is Record<string, unknown> { return typeof value === "object" && value !== null && !Array.isArray(value);} function markSeen(value: object, seen: WeakSet<object>): boolean { if (seen.has(value)) { return true; } seen.add(value); return false;} function normalizeJsonValue(value: unknown, seen: WeakSet<object> = new WeakSet()): unknown { if ( value === null || value === undefined || typeof value === "string" || typeof value === "number" || typeof value === "boolean" ) { return value ?? null; } if (typeof value === "bigint") { return value.toString(); } if (value instanceof Date) { return Number.isNaN(value.getTime()) ? "Invalid Date" : value.toISOString(); } if (value instanceof Error) { return { name: value.name, message: value.message, ...(value.stack ? { stack: value.stack } : {}), }; } if (Array.isArray(value)) { if (markSeen(value, seen)) { return "[Circular]"; } return value.map((entry) => normalizeJsonValue(entry, seen)); } if (value instanceof Map) { if (markSeen(value, seen)) { return "[Circular]"; } return Object.fromEntries( Array.from(value.entries(), ([key, entryValue]) => [ String(key), normalizeJsonValue(entryValue, seen), ]), ); } if (value instanceof Set) { if (markSeen(value, seen)) { return "[Circular]"; } return Array.from(value.values(), (entry) => normalizeJsonValue(entry, seen)); } if (!isPlainObject(value)) { return String(value); } if (markSeen(value, seen)) { return "[Circular]"; } return Object.fromEntries( Object.entries(value).map(([key, entryValue]) => [key, normalizeJsonValue(entryValue, seen)]), );} export function compactTraceAttributes( attributes: Readonly<Record<string, unknown>>,): TraceAttributes { const entries: Array<[string, unknown]> = []; for (const [key, value] of Object.entries(attributes)) { if (value !== undefined) { entries.push([key, normalizeJsonValue(value)]); } } return Object.fromEntries(entries);} function formatTraceExit(exit: Exit.Exit<unknown, unknown>): EffectTraceRecord["exit"] { if (ExitRuntime.isSuccess(exit)) { return { _tag: "Success" }; } if (Cause.hasInterruptsOnly(exit.cause)) { return { _tag: "Interrupted", cause: Cause.pretty(exit.cause), }; } return { _tag: "Failure", cause: Cause.pretty(exit.cause), };} const TRACE_ATTRIBUTE_MAX_LENGTH = 500;const TRACE_ATTRIBUTE_TRUNCATED_LENGTH = 200;const TRACE_ATTRIBUTE_TRUNCATION_SUFFIX = "…[truncated]";const ALWAYS_TRUNCATED_TRACE_ATTRIBUTES: ReadonlySet<string> = new Set(["db.query.text"]); // Clamps strings nested inside already-normalized attribute values (arrays and// plain objects from normalizeJsonValue, e.g. an Error's `stack`). Returns the// input reference when nothing was clamped.function truncateNestedValue(value: unknown): unknown { if (typeof value === "string") { return value.length <= TRACE_ATTRIBUTE_MAX_LENGTH ? value : `${value.slice(0, TRACE_ATTRIBUTE_MAX_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`; } if (Array.isArray(value)) { const truncated = value.map(truncateNestedValue); return truncated.some((entry, index) => entry !== value[index]) ? truncated : value; } if (isPlainObject(value)) { let truncated: Record<string, unknown> | undefined; for (const [key, entry] of Object.entries(value)) { const next = truncateNestedValue(entry); if (next === entry) continue; truncated ??= { ...value }; truncated[key] = next; } return truncated ?? value; } return value;} /** * Clamps oversized attribute values on the serialized trace record so the file * sink stays small, including strings nested inside arrays and objects (e.g. * error stacks). Returns a new record when anything was clamped; never * mutates the input (the live span's attributes are shared with other tracers). */export function truncateTraceAttributes(attributes: TraceAttributes): TraceAttributes { let truncated: Record<string, unknown> | undefined; for (const [key, value] of Object.entries(attributes)) { if (typeof value === "string" && ALWAYS_TRUNCATED_TRACE_ATTRIBUTES.has(key)) { if (value.length <= TRACE_ATTRIBUTE_TRUNCATED_LENGTH) continue; truncated ??= { ...attributes }; truncated[key] = `${value.slice(0, TRACE_ATTRIBUTE_TRUNCATED_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`; continue; } const next = truncateNestedValue(value); if (next === value) continue; truncated ??= { ...attributes }; truncated[key] = next; } return truncated ?? attributes;} function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord { const status = span.status as Extract<Tracer.SpanStatus, { _tag: "Ended" }>; const parentSpanId = Option.getOrUndefined(span.parent)?.spanId; return { type: "effect-span", name: span.name, traceId: span.traceId, spanId: span.spanId, ...(parentSpanId ? { parentSpanId } : {}), sampled: span.sampled, kind: span.kind, startTimeUnixNano: String(status.startTime), endTimeUnixNano: String(status.endTime), durationMs: Number(status.endTime - status.startTime) / 1_000_000, attributes: truncateTraceAttributes( compactTraceAttributes(Object.fromEntries(span.attributes)), ), events: span.events.map(([name, startTime, attributes]) => ({ name, timeUnixNano: String(startTime), attributes: truncateTraceAttributes(compactTraceAttributes(attributes)), })), links: span.links.map((link) => ({ traceId: link.span.traceId, spanId: link.span.spanId, attributes: truncateTraceAttributes(compactTraceAttributes(link.attributes)), })), exit: formatTraceExit(status.exit), };} export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) { const sink = new RotatingFileSink({ filePath: options.filePath, maxBytes: options.maxBytes, maxFiles: options.maxFiles, throwOnError: true, }); let buffer: Array<string> = []; let pendingFlushStats: TraceSinkFlushStats = { logicalWriteBytes: 0, count: 0, durationMs: 0, }; const flushUnsafe = () => { if (buffer.length === 0) { return; } const records = buffer; buffer = []; let persistedCount = 0; while (persistedCount < records.length) { const firstRecordBytes = textEncoder.encode(records[persistedCount]).byteLength; if (firstRecordBytes > options.maxBytes) { persistedCount += 1; continue; } let nextIndex = persistedCount + 1; let chunkBytes = firstRecordBytes; while (nextIndex < records.length) { const nextRecordBytes = textEncoder.encode(records[nextIndex]).byteLength; if (chunkBytes + nextRecordBytes > options.maxBytes) break; chunkBytes += nextRecordBytes; nextIndex += 1; } const chunk = records.slice(persistedCount, nextIndex).join(""); const startedAt = performance.now(); try { sink.write(chunk); } catch { buffer.unshift(...records.slice(persistedCount)); return; } pendingFlushStats = { logicalWriteBytes: pendingFlushStats.logicalWriteBytes + chunkBytes, count: pendingFlushStats.count + nextIndex - persistedCount, durationMs: pendingFlushStats.durationMs + Math.max(0, performance.now() - startedAt), }; persistedCount = nextIndex; } }; const flush = Effect.sync(() => { flushUnsafe(); const stats = pendingFlushStats; pendingFlushStats = { logicalWriteBytes: 0, count: 0, durationMs: 0, }; return stats; }).pipe( Effect.flatMap((stats) => stats.count > 0 && options.onFlush ? options.onFlush(stats).pipe(Effect.ignore) : Effect.void, ), Effect.withTracerEnabled(false), ); yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore)); yield* Effect.forkScoped( Effect.sleep(`${options.batchWindowMs} millis`).pipe(Effect.andThen(flush), Effect.forever), ); return { filePath: options.filePath, push(record) { try { buffer.push(`${JSON.stringify(record)}\n`); if (buffer.length >= FLUSH_BUFFER_THRESHOLD) { flushUnsafe(); } } catch { return; } }, flush, close: () => flush, } satisfies TraceSink;}); class LocalFileSpan implements Tracer.Span { readonly _tag = "Span"; readonly name: string; readonly spanId: string; readonly traceId: string; readonly parent: Option.Option<Tracer.AnySpan>; readonly annotations: Tracer.Span["annotations"]; readonly links: Array<Tracer.SpanLink>; readonly sampled: boolean; readonly kind: Tracer.SpanKind; status: Tracer.SpanStatus; attributes: Map<string, unknown>; events: Array<[name: string, startTime: bigint, attributes: Record<string, unknown>]>; private readonly delegate: Tracer.Span; private readonly push: (record: EffectTraceRecord) => void; constructor( options: Parameters<Tracer.Tracer["span"]>[0], delegate: Tracer.Span, push: (record: EffectTraceRecord) => void, ) { this.delegate = delegate; this.push = push; this.name = delegate.name; this.spanId = delegate.spanId; this.traceId = delegate.traceId; this.parent = options.parent; this.annotations = options.annotations; this.links = [...options.links]; this.sampled = delegate.sampled; this.kind = delegate.kind; this.status = { _tag: "Started", startTime: options.startTime, }; this.attributes = new Map(); this.events = []; } end(endTime: bigint, exit: Exit.Exit<unknown, unknown>): void { this.status = { _tag: "Ended", startTime: this.status.startTime, endTime, exit, }; this.delegate.end(endTime, exit); if (this.sampled) { this.push(spanToTraceRecord(this)); } } attribute(key: string, value: unknown): void { this.attributes.set(key, value); this.delegate.attribute(key, value); } event(name: string, startTime: bigint, attributes?: Record<string, unknown>): void { const nextAttributes = attributes ?? {}; this.events.push([name, startTime, nextAttributes]); this.delegate.event(name, startTime, nextAttributes); } addLinks(links: ReadonlyArray<Tracer.SpanLink>): void { this.links.push(...links); this.delegate.addLinks(links);Lines 1–500 of 685. Load more
import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import type * as Exit from "effect/Exit"; import * as ExitRuntime from "effect/Exit"; import * as Option from "effect/Option"; import * as Tracer from "effect/Tracer"; import { OtlpResource, OtlpTracer } from "effect/unstable/observability"; import { RotatingFileSink } from "./logging.ts"; const FLUSH_BUFFER_THRESHOLD = 256; const textEncoder = new TextEncoder(); export type TraceAttributes = Readonly<Record<string, unknown>>; export interface TraceRecordEvent { readonly name: string; readonly timeUnixNano: string; readonly attributes: Readonly<Record<string, unknown>>; } export interface TraceRecordLink { readonly traceId: string; readonly spanId: string; readonly attributes: Readonly<Record<string, unknown>>; } interface BaseTraceRecord { readonly name: string; readonly kind: string; readonly traceId: string; readonly spanId: string; readonly parentSpanId?: string; readonly sampled: boolean; readonly startTimeUnixNano: string; readonly endTimeUnixNano: string; readonly durationMs: number; readonly attributes: Readonly<Record<string, unknown>>; readonly events: ReadonlyArray<TraceRecordEvent>; readonly links: ReadonlyArray<TraceRecordLink>; } export interface EffectTraceRecord extends BaseTraceRecord { readonly type: "effect-span"; readonly exit: | { readonly _tag: "Success"; } | { readonly _tag: "Interrupted"; readonly cause: string; } | { readonly _tag: "Failure"; readonly cause: string; }; } export interface OtlpTraceRecord extends BaseTraceRecord { readonly type: "otlp-span"; readonly resourceAttributes: Readonly<Record<string, unknown>>; readonly scope: Readonly<{ readonly name?: string; readonly version?: string; readonly attributes: Readonly<Record<string, unknown>>; }>; readonly status?: | { readonly code?: string; readonly message?: string; } | undefined; } export type TraceRecord = EffectTraceRecord | OtlpTraceRecord; function isStructuralTag(value: unknown): value is string { return ( typeof value === "string" && value.length > 0 && value.length <= 128 && /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value) ); } export function errorTag(error: unknown): string { try { if (typeof error === "object" && error !== null && "_tag" in error) { return isStructuralTag(error._tag) ? error._tag : "TaggedError"; } if (error instanceof Error) { return isStructuralTag(error.name) ? error.name : "Error"; } } catch { return "UnknownError"; } return typeof error; } export function causeErrorTag(cause: Cause.Cause<unknown>): string { const failure = Cause.findErrorOption(cause); if (Option.isSome(failure)) { return errorTag(failure.value); } return cause.reasons[0]?._tag ?? "Empty"; } export interface TraceSinkOptions { readonly filePath: string; readonly maxBytes: number; readonly maxFiles: number; readonly batchWindowMs: number; readonly onFlush?: (stats: TraceSinkFlushStats) => Effect.Effect<void>; } export interface TraceSinkFlushStats { readonly logicalWriteBytes: number; readonly count: number; readonly durationMs: number; } export interface TraceSink { readonly filePath: string; push: (record: TraceRecord) => void; flush: Effect.Effect<void>; close: () => Effect.Effect<void>; } export interface LocalFileTracerOptions extends TraceSinkOptions { readonly delegate?: Tracer.Tracer; readonly sink?: TraceSink; } type OtlpSpan = OtlpTracer.ScopeSpan["spans"][number]; type OtlpSpanEvent = OtlpSpan["events"][number]; type OtlpSpanLink = OtlpSpan["links"][number]; type OtlpSpanStatus = OtlpSpan["status"]; interface SerializableSpan { readonly name: string; readonly traceId: string; readonly spanId: string; readonly parent: Option.Option<Tracer.AnySpan>; readonly status: Tracer.SpanStatus; readonly sampled: boolean; readonly kind: Tracer.SpanKind; readonly attributes: ReadonlyMap<string, unknown>; readonly links: ReadonlyArray<Tracer.SpanLink>; readonly events: ReadonlyArray< readonly [name: string, startTime: bigint, attributes: Record<string, unknown>] >; } function isPlainObject(value: unknown): value is Record<string, unknown> { return typeof value === "object" && value !== null && !Array.isArray(value); } function markSeen(value: object, seen: WeakSet<object>): boolean { if (seen.has(value)) { return true; } seen.add(value); return false; } function normalizeJsonValue(value: unknown, seen: WeakSet<object> = new WeakSet()): unknown { if ( value === null || value === undefined || typeof value === "string" || typeof value === "number" || typeof value === "boolean" ) { return value ?? null; } if (typeof value === "bigint") { return value.toString(); } if (value instanceof Date) { return Number.isNaN(value.getTime()) ? "Invalid Date" : value.toISOString(); } if (value instanceof Error) { return { name: value.name, message: value.message, ...(value.stack ? { stack: value.stack } : {}), }; } if (Array.isArray(value)) { if (markSeen(value, seen)) { return "[Circular]"; } return value.map((entry) => normalizeJsonValue(entry, seen)); } if (value instanceof Map) { if (markSeen(value, seen)) { return "[Circular]"; } return Object.fromEntries( Array.from(value.entries(), ([key, entryValue]) => [ String(key), normalizeJsonValue(entryValue, seen), ]), ); } if (value instanceof Set) { if (markSeen(value, seen)) { return "[Circular]"; } return Array.from(value.values(), (entry) => normalizeJsonValue(entry, seen)); } if (!isPlainObject(value)) { return String(value); } if (markSeen(value, seen)) { return "[Circular]"; } return Object.fromEntries( Object.entries(value).map(([key, entryValue]) => [key, normalizeJsonValue(entryValue, seen)]), ); } export function compactTraceAttributes( attributes: Readonly<Record<string, unknown>>, ): TraceAttributes { const entries: Array<[string, unknown]> = []; for (const [key, value] of Object.entries(attributes)) { if (value !== undefined) { entries.push([key, normalizeJsonValue(value)]); } } return Object.fromEntries(entries); } function formatTraceExit(exit: Exit.Exit<unknown, unknown>): EffectTraceRecord["exit"] { if (ExitRuntime.isSuccess(exit)) { return { _tag: "Success" }; } if (Cause.hasInterruptsOnly(exit.cause)) { return { _tag: "Interrupted", cause: Cause.pretty(exit.cause), }; } return { _tag: "Failure", cause: Cause.pretty(exit.cause), }; } const TRACE_ATTRIBUTE_MAX_LENGTH = 500; const TRACE_ATTRIBUTE_TRUNCATED_LENGTH = 200; const TRACE_ATTRIBUTE_TRUNCATION_SUFFIX = "…[truncated]"; const ALWAYS_TRUNCATED_TRACE_ATTRIBUTES: ReadonlySet<string> = new Set(["db.query.text"]); // Clamps strings nested inside already-normalized attribute values (arrays and // plain objects from normalizeJsonValue, e.g. an Error's `stack`). Returns the // input reference when nothing was clamped. function truncateNestedValue(value: unknown): unknown { if (typeof value === "string") { return value.length <= TRACE_ATTRIBUTE_MAX_LENGTH ? value : `${value.slice(0, TRACE_ATTRIBUTE_MAX_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`; } if (Array.isArray(value)) { const truncated = value.map(truncateNestedValue); return truncated.some((entry, index) => entry !== value[index]) ? truncated : value; } if (isPlainObject(value)) { let truncated: Record<string, unknown> | undefined; for (const [key, entry] of Object.entries(value)) { const next = truncateNestedValue(entry); if (next === entry) continue; truncated ??= { ...value }; truncated[key] = next; } return truncated ?? value; } return value; } /** * Clamps oversized attribute values on the serialized trace record so the file * sink stays small, including strings nested inside arrays and objects (e.g. * error stacks). Returns a new record when anything was clamped; never * mutates the input (the live span's attributes are shared with other tracers). */ export function truncateTraceAttributes(attributes: TraceAttributes): TraceAttributes { let truncated: Record<string, unknown> | undefined; for (const [key, value] of Object.entries(attributes)) { if (typeof value === "string" && ALWAYS_TRUNCATED_TRACE_ATTRIBUTES.has(key)) { if (value.length <= TRACE_ATTRIBUTE_TRUNCATED_LENGTH) continue; truncated ??= { ...attributes }; truncated[key] = `${value.slice(0, TRACE_ATTRIBUTE_TRUNCATED_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`; continue; } const next = truncateNestedValue(value); if (next === value) continue; truncated ??= { ...attributes }; truncated[key] = next; } return truncated ?? attributes; } function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord { const status = span.status as Extract<Tracer.SpanStatus, { _tag: "Ended" }>; const parentSpanId = Option.getOrUndefined(span.parent)?.spanId; return { type: "effect-span", name: span.name, traceId: span.traceId, spanId: span.spanId, ...(parentSpanId ? { parentSpanId } : {}), sampled: span.sampled, kind: span.kind, startTimeUnixNano: String(status.startTime), endTimeUnixNano: String(status.endTime), durationMs: Number(status.endTime - status.startTime) / 1_000_000, attributes: truncateTraceAttributes( compactTraceAttributes(Object.fromEntries(span.attributes)), ), events: span.events.map(([name, startTime, attributes]) => ({ name, timeUnixNano: String(startTime), attributes: truncateTraceAttributes(compactTraceAttributes(attributes)), })), links: span.links.map((link) => ({ traceId: link.span.traceId, spanId: link.span.spanId, attributes: truncateTraceAttributes(compactTraceAttributes(link.attributes)), })), exit: formatTraceExit(status.exit), }; } export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) { const sink = new RotatingFileSink({ filePath: options.filePath, maxBytes: options.maxBytes, maxFiles: options.maxFiles, throwOnError: true, }); let buffer: Array<string> = []; let pendingFlushStats: TraceSinkFlushStats = { logicalWriteBytes: 0, count: 0, durationMs: 0, }; const flushUnsafe = () => { if (buffer.length === 0) { return; } const records = buffer; buffer = []; let persistedCount = 0; while (persistedCount < records.length) { const firstRecordBytes = textEncoder.encode(records[persistedCount]).byteLength; if (firstRecordBytes > options.maxBytes) { persistedCount += 1; continue; } let nextIndex = persistedCount + 1; let chunkBytes = firstRecordBytes; while (nextIndex < records.length) { const nextRecordBytes = textEncoder.encode(records[nextIndex]).byteLength; if (chunkBytes + nextRecordBytes > options.maxBytes) break; chunkBytes += nextRecordBytes; nextIndex += 1; } const chunk = records.slice(persistedCount, nextIndex).join(""); const startedAt = performance.now(); try { sink.write(chunk); } catch { buffer.unshift(...records.slice(persistedCount)); return; } pendingFlushStats = { logicalWriteBytes: pendingFlushStats.logicalWriteBytes + chunkBytes, count: pendingFlushStats.count + nextIndex - persistedCount, durationMs: pendingFlushStats.durationMs + Math.max(0, performance.now() - startedAt), }; persistedCount = nextIndex; } }; const flush = Effect.sync(() => { flushUnsafe(); const stats = pendingFlushStats; pendingFlushStats = { logicalWriteBytes: 0, count: 0, durationMs: 0, }; return stats; }).pipe( Effect.flatMap((stats) => stats.count > 0 && options.onFlush ? options.onFlush(stats).pipe(Effect.ignore) : Effect.void, ), Effect.withTracerEnabled(false), ); yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore)); yield* Effect.forkScoped( Effect.sleep(`${options.batchWindowMs} millis`).pipe(Effect.andThen(flush), Effect.forever), ); return { filePath: options.filePath, push(record) { try { buffer.push(`${JSON.stringify(record)}\n`); if (buffer.length >= FLUSH_BUFFER_THRESHOLD) { flushUnsafe(); } } catch { return; } }, flush, close: () => flush, } satisfies TraceSink; }); class LocalFileSpan implements Tracer.Span { readonly _tag = "Span"; readonly name: string; readonly spanId: string; readonly traceId: string; readonly parent: Option.Option<Tracer.AnySpan>; readonly annotations: Tracer.Span["annotations"]; readonly links: Array<Tracer.SpanLink>; readonly sampled: boolean; readonly kind: Tracer.SpanKind; status: Tracer.SpanStatus; attributes: Map<string, unknown>; events: Array<[name: string, startTime: bigint, attributes: Record<string, unknown>]>; private readonly delegate: Tracer.Span; private readonly push: (record: EffectTraceRecord) => void; constructor( options: Parameters<Tracer.Tracer["span"]>[0], delegate: Tracer.Span, push: (record: EffectTraceRecord) => void, ) { this.delegate = delegate; this.push = push; this.name = delegate.name; this.spanId = delegate.spanId; this.traceId = delegate.traceId; this.parent = options.parent; this.annotations = options.annotations; this.links = [...options.links]; this.sampled = delegate.sampled; this.kind = delegate.kind; this.status = { _tag: "Started", startTime: options.startTime, }; this.attributes = new Map(); this.events = []; } end(endTime: bigint, exit: Exit.Exit<unknown, unknown>): void { this.status = { _tag: "Ended", startTime: this.status.startTime, endTime, exit, }; this.delegate.end(endTime, exit); if (this.sampled) { this.push(spanToTraceRecord(this)); } } attribute(key: string, value: unknown): void { this.attributes.set(key, value); this.delegate.attribute(key, value); } event(name: string, startTime: bigint, attributes?: Record<string, unknown>): void { const nextAttributes = attributes ?? {}; this.events.push([name, startTime, nextAttributes]); this.delegate.event(name, startTime, nextAttributes); } addLinks(links: ReadonlyArray<Tracer.SpanLink>): void { this.links.push(...links); this.delegate.addLinks(links);