1import type {
2 ServerTraceDiagnosticsErrorKind,
3 ServerTraceDiagnosticsFailureSummary,
4 ServerTraceDiagnosticsLogEvent,
5 ServerTraceDiagnosticsRecentFailure,
6 ServerTraceDiagnosticsResult,
7 ServerTraceDiagnosticsSpanOccurrence,
8 ServerTraceDiagnosticsSpanSummary,
9} from "@t3tools/contracts";
10import * as Context from "effect/Context";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import * as FileSystem from "effect/FileSystem";
14import * as Layer from "effect/Layer";
15import * as Option from "effect/Option";
16import * as PlatformError from "effect/PlatformError";
17import * as Result from "effect/Result";
18import * as Schema from "effect/Schema";
19import * as Stream from "effect/Stream";
20
21interface TraceRecordLike {
22 readonly name?: unknown;
23 readonly traceId?: unknown;
24 readonly spanId?: unknown;
25 readonly startTimeUnixNano?: unknown;
26 readonly endTimeUnixNano?: unknown;
27 readonly durationMs?: unknown;
28 readonly exit?: unknown;
29 readonly events?: unknown;
30}
31
32interface TraceEventLike {
33 readonly name?: unknown;
34 readonly timeUnixNano?: unknown;
35 readonly attributes?: unknown;
36}
37
38export interface TraceDiagnosticsOptions {
39 readonly traceFilePath: string;
40 readonly maxFiles: number;
41 readonly slowSpanThresholdMs?: number;
42 readonly readAt?: DateTime.Utc;
43}
44
45export class TraceFileReadError extends Schema.TaggedError<TraceFileReadError>()(
46 "TraceFileReadError",
47 {
48 traceFilePath: Schema.String,
49 causeTag: Schema.String,
50 cause: Schema.Defect(),
51 },
52) {
53 override get message(): string {
54 return `Failed to read local trace file '${this.traceFilePath}'.`;
55 }
56}
57
58export class TraceDiagnostics extends Context.Service<
59 TraceDiagnostics,
60 {
61 readonly read: (
62 options: TraceDiagnosticsOptions,
63 ) => Effect.Effect<ServerTraceDiagnosticsResult>;
64 }
65>()("t3/diagnostics/TraceDiagnostics") {}
66
67interface TraceDiagnosticsErrorSummary {
68 readonly kind: ServerTraceDiagnosticsErrorKind;
69 readonly message: string;
70}
71
72const DEFAULT_SLOW_SPAN_THRESHOLD_MS = 1_000;
73const TOP_LIMIT = 10;
74const RECENT_LIMIT = 20;
75
76/** The trace file and its rotated backups, oldest first. */
77export function toRotatedTracePaths(
78 traceFilePath: string,
79 maxFiles: number,
80): ReadonlyArray<string> {
81 const backupCount = Math.max(0, Math.floor(maxFiles));
82 const backups = Array.from(
83 { length: backupCount },
84 (_, index) => `${traceFilePath}.${backupCount - index}`,
85 );
86 return [...backups, traceFilePath];
87}
88
89function isRecordObject(value: unknown): value is TraceRecordLike {
90 return typeof value === "object" && value !== null;
91}
92
93function toStringValue(value: unknown): string | null {
94 return typeof value === "string" && value.trim().length > 0 ? value : null;
95}
96
97function toNumberValue(value: unknown): number | null {
98 return typeof value === "number" && Number.isFinite(value) ? value : null;
99}
100
101function unixNanoToDateTime(value: unknown): DateTime.Utc | null {
102 const text = toStringValue(value);
103 if (!text) return null;
104 try {
105 const millis = Number(BigInt(text) / 1_000_000n);
106 return Option.getOrNull(DateTime.make(millis));
107 } catch {
108 return null;
109 }
110}
111
112function readExitTag(exit: unknown): string | null {
113 if (!isRecordObject(exit) || !("_tag" in exit)) return null;
114 return toStringValue(exit._tag);
115}
116
117function readExitCause(exit: unknown): string {
118 if (!isRecordObject(exit) || !("cause" in exit)) return "Failure";
119 return toStringValue(exit.cause)?.trim() ?? "Failure";
120}
121
122function isTraceEvent(value: unknown): value is TraceEventLike {
123 return typeof value === "object" && value !== null;
124}
125
126function readEventAttributes(event: TraceEventLike): Readonly<Record<string, unknown>> {
127 return typeof event.attributes === "object" && event.attributes !== null
128 ? (event.attributes as Readonly<Record<string, unknown>>)
129 : {};
130}
131
132function makeEmptyDiagnostics(input: {
133 readonly traceFilePath: string;
134 readonly scannedFilePaths: ReadonlyArray<string>;
135 readonly readAt: DateTime.Utc;
136 readonly slowSpanThresholdMs: number;
137 readonly error?: TraceDiagnosticsErrorSummary;
138 readonly partialFailure?: boolean;
139}): ServerTraceDiagnosticsResult {
140 return {
141 traceFilePath: input.traceFilePath,
142 scannedFilePaths: [...input.scannedFilePaths],
143 readAt: input.readAt,
144 recordCount: 0,
145 parseErrorCount: 0,
146 firstSpanAt: Option.none(),
147 lastSpanAt: Option.none(),
148 failureCount: 0,
149 interruptionCount: 0,
150 slowSpanThresholdMs: input.slowSpanThresholdMs,
151 slowSpanCount: 0,
152 logLevelCounts: {},
153 topSpansByCount: [],
154 slowestSpans: [],
155 commonFailures: [],
156 latestFailures: [],
157 latestWarningAndErrorLogs: [],
158 partialFailure: input.partialFailure ? Option.some(true) : Option.none(),
159 error: Option.fromNullishOr(input.error),
160 };
161}
162
163function isNotFoundError(error: PlatformError.PlatformError): boolean {
164 return error.reason._tag === "NotFound";
165}
166
167/**
168 * Adds `item` to `items`, which stays sorted by `order` and holds at most
169 * `limit` entries. Same result as a stable sort and slice over every item, but
170 * memory stays bounded however many items stream in.
171 */
172function insertBounded<A>(
173 items: A[],
174 item: A,
175 limit: number,
176 order: (left: A, right: A) => number,
177): void {
178 if (items.length >= limit && order(item, items[items.length - 1]!) >= 0) {
179 return;
180 }
181
182 items.push(item);
183 items.sort(order);
184 if (items.length > limit) {
185 items.length = limit;
186 }
187}
188
189const slowestFirst = (
190 left: ServerTraceDiagnosticsSpanOccurrence,
191 right: ServerTraceDiagnosticsSpanOccurrence,
192) => right.durationMs - left.durationMs;
193
194const latestEndedFirst = (
195 left: ServerTraceDiagnosticsRecentFailure,
196 right: ServerTraceDiagnosticsRecentFailure,
197) => DateTime.toEpochMillis(right.endedAt) - DateTime.toEpochMillis(left.endedAt);
198
199const latestSeenFirst = (
200 left: ServerTraceDiagnosticsLogEvent,
201 right: ServerTraceDiagnosticsLogEvent,
202) => DateTime.toEpochMillis(right.seenAt) - DateTime.toEpochMillis(left.seenAt);
203
204/**
205 * Folds trace NDJSON into diagnostics. Call `addLine` once per line as the
206 * rotated files stream in, then `finish` for the result.
207 */
208export function makeTraceDiagnosticsAggregator(
209 slowSpanThresholdMs = DEFAULT_SLOW_SPAN_THRESHOLD_MS,
210) {
211 let parseErrorCount = 0;
212 let recordCount = 0;
213 let failureCount = 0;
214 let interruptionCount = 0;
215 let slowSpanCount = 0;
216 let firstSpanAt: DateTime.Utc | null = null;
217 let lastSpanAt: DateTime.Utc | null = null;
218
219 const spansByName = new Map<
220 string,
221 { count: number; failureCount: number; totalDurationMs: number; maxDurationMs: number }
222 >();
223 const failuresByKey = new Map<string, ServerTraceDiagnosticsFailureSummary>();
224 const latestFailures: ServerTraceDiagnosticsRecentFailure[] = [];
225 const slowestSpans: ServerTraceDiagnosticsSpanOccurrence[] = [];
226 const latestWarningAndErrorLogs: ServerTraceDiagnosticsLogEvent[] = [];
227 const logLevelCounts: Record<string, number> = {};
228
229 const addLine = (line: string) => {
230 if (line.trim().length === 0) return;
231
232 let parsed: unknown;
233 try {
234 parsed = JSON.parse(line);
235 } catch {
236 parseErrorCount += 1;
237 return;
238 }
239
240 if (!isRecordObject(parsed)) {
241 parseErrorCount += 1;
242 return;
243 }
244
245 const name = toStringValue(parsed.name);
246 const traceId = toStringValue(parsed.traceId);
247 const spanId = toStringValue(parsed.spanId);
248 const durationMs = toNumberValue(parsed.durationMs);
249 const endedAt = unixNanoToDateTime(parsed.endTimeUnixNano);
250 const startedAt = unixNanoToDateTime(parsed.startTimeUnixNano);
251
252 if (!name || !traceId || !spanId || durationMs === null || !endedAt) {
253 parseErrorCount += 1;
254 return;
255 }
256
257 recordCount += 1;
258 firstSpanAt =
259 startedAt && (firstSpanAt === null || DateTime.isLessThan(startedAt, firstSpanAt))
260 ? startedAt
261 : firstSpanAt;
262 lastSpanAt =
263 lastSpanAt === null || DateTime.isGreaterThan(endedAt, lastSpanAt) ? endedAt : lastSpanAt;
264
265 const exitTag = readExitTag(parsed.exit);
266 const isFailure = exitTag === "Failure";
267 const isInterrupted = exitTag === "Interrupted";
268 if (isFailure) failureCount += 1;
269 if (isInterrupted) interruptionCount += 1;
270
271 const spanSummary = spansByName.get(name) ?? {
272 count: 0,
273 failureCount: 0,
274 totalDurationMs: 0,
275 maxDurationMs: 0,
276 };
277 spanSummary.count += 1;
278 spanSummary.totalDurationMs += durationMs;
279 spanSummary.maxDurationMs = Math.max(spanSummary.maxDurationMs, durationMs);
280 if (isFailure) spanSummary.failureCount += 1;
281 spansByName.set(name, spanSummary);
282
283 const spanItem = { name, durationMs, endedAt, traceId, spanId };
284 if (durationMs >= slowSpanThresholdMs) {
285 slowSpanCount += 1;
286 }
287 insertBounded(slowestSpans, spanItem, TOP_LIMIT, slowestFirst);
288
289 if (isFailure) {
290 const cause = readExitCause(parsed.exit);
291 insertBounded(latestFailures, { ...spanItem, cause }, RECENT_LIMIT, latestEndedFirst);
292
293 const failureKey = `${name}\0${cause}`;
294 const existing = failuresByKey.get(failureKey);
295 const isLatestFailure = !existing || DateTime.isGreaterThan(endedAt, existing.lastSeenAt);
296 failuresByKey.set(failureKey, {
297 name,
298 cause,
299 count: (existing?.count ?? 0) + 1,
300 lastSeenAt: isLatestFailure ? endedAt : existing!.lastSeenAt,
301 traceId: isLatestFailure ? traceId : existing!.traceId,
302 spanId: isLatestFailure ? spanId : existing!.spanId,
303 });
304 }
305
306 if (Array.isArray(parsed.events)) {
307 for (const rawEvent of parsed.events) {
308 if (!isTraceEvent(rawEvent)) continue;
309 const attributes = readEventAttributes(rawEvent);
310 const level = toStringValue(attributes["effect.logLevel"]);
311 if (!level) continue;
312
313 logLevelCounts[level] = (logLevelCounts[level] ?? 0) + 1;
314 const normalizedLevel = level.toLowerCase();
315 if (
316 normalizedLevel !== "warning" &&
317 normalizedLevel !== "warn" &&
318 normalizedLevel !== "error" &&
319 normalizedLevel !== "fatal"
320 ) {
321 continue;
322 }
323
324 const seenAt = unixNanoToDateTime(rawEvent.timeUnixNano) ?? endedAt;
325 const message = toStringValue(rawEvent.name)?.trim() ?? "Log event";
326 insertBounded(
327 latestWarningAndErrorLogs,
328 { spanName: name, level, message, seenAt, traceId, spanId },
329 RECENT_LIMIT,
330 latestSeenFirst,
331 );
332 }
333 }
334 };
335
336 const finish = (input: {
337 readonly traceFilePath: string;
338 readonly scannedFilePaths: ReadonlyArray<string>;
339 readonly readAt: DateTime.Utc;
340 readonly error?: TraceDiagnosticsErrorSummary;
341 readonly partialFailure?: boolean;
342 }): ServerTraceDiagnosticsResult => {
343 const topSpansByCount: ServerTraceDiagnosticsSpanSummary[] = [...spansByName.entries()]
344 .map(([name, span]) => ({
345 name,
346 count: span.count,
347 failureCount: span.failureCount,
348 totalDurationMs: span.totalDurationMs,
349 averageDurationMs: span.count > 0 ? span.totalDurationMs / span.count : 0,
350 maxDurationMs: span.maxDurationMs,
351 }))
352 .toSorted(
353 (left, right) => right.count - left.count || right.maxDurationMs - left.maxDurationMs,
354 )
355 .slice(0, TOP_LIMIT);
356
357 return {
358 traceFilePath: input.traceFilePath,
359 scannedFilePaths: input.scannedFilePaths,
360 readAt: input.readAt,
361 recordCount,
362 parseErrorCount,
363 firstSpanAt: Option.fromNullishOr(firstSpanAt),
364 lastSpanAt: Option.fromNullishOr(lastSpanAt),
365 failureCount,
366 interruptionCount,
367 slowSpanThresholdMs,
368 slowSpanCount,
369 logLevelCounts,
370 topSpansByCount,
371 slowestSpans,
372 commonFailures: [...failuresByKey.values()]
373 .toSorted(
374 (left, right) =>
375 right.count - left.count ||
376 DateTime.toEpochMillis(right.lastSeenAt) - DateTime.toEpochMillis(left.lastSeenAt),
377 )
378 .slice(0, TOP_LIMIT),
379 latestFailures,
380 latestWarningAndErrorLogs,
381 partialFailure: input.partialFailure ? Option.some(true) : Option.none(),
382 error: Option.fromNullishOr(input.error),
383 };
384 };
385
386 return { addLine, finish };
387}
388
389/**
390 * Feeds each line of one trace file to `onLine`, streaming so only one chunk of
391 * text is in memory at a time. Succeeds with false when the file does not exist.
392 */
393export function streamTraceFileLines(
394 fileSystem: FileSystem.FileSystem,
395 path: string,
396 onLine: (line: string) => void,
397): Effect.Effect<boolean, TraceFileReadError> {
398 return fileSystem.stream(path).pipe(
399 Stream.decodeText,
400 Stream.splitLines,
401 Stream.runForEachArray((lines) => Effect.sync(() => lines.forEach(onLine))),
402 Effect.as(true),
403 Effect.catchTags({
404 PlatformError: (cause) =>
405 isNotFoundError(cause)
406 ? Effect.succeed(false)
407 : Effect.fail(
408 new TraceFileReadError({
409 traceFilePath: path,
410 causeTag: cause.reason._tag,
411 cause,
412 }),
413 ),
414 }),
415 );
416}
417
418/** @public Service construction is part of the canonical Effect module API. */
419export const make = Effect.gen(function* () {
420 const fileSystem = yield* FileSystem.FileSystem;
421
422 const read: TraceDiagnostics["Service"]["read"] = Effect.fn("TraceDiagnostics.read")(
423 function* (options) {
424 const readAt = options.readAt ?? (yield* DateTime.now);
425 const slowSpanThresholdMs = options.slowSpanThresholdMs ?? DEFAULT_SLOW_SPAN_THRESHOLD_MS;
426 const paths = toRotatedTracePaths(options.traceFilePath, options.maxFiles);
427 const aggregator = makeTraceDiagnosticsAggregator(slowSpanThresholdMs);
428 const results = yield* Effect.forEach(
429 paths,
430 (path) =>
431 streamTraceFileLines(fileSystem, path, aggregator.addLine).pipe(
432 Effect.tapError((cause) =>
433 Effect.logWarning("Failed to read local trace file.").pipe(
434 Effect.annotateLogs({
435 traceFilePath: cause.traceFilePath,
436 errorTag: cause._tag,
437 causeTag: cause.causeTag,
438 }),
439 ),
440 ),
441 Effect.result,
442 ),
443 // Every file feeds one aggregator, so read them one at a time, oldest first.
444 { concurrency: 1 },
445 );
446 const foundFile = results.some((result) => Result.isSuccess(result) && result.success);
447 const readFailure = results.find(Result.isFailure);
448 const readFailureError = readFailure
449 ? ({
450 kind: "trace-file-read-failed",
451 message: readFailure.failure.message,
452 } satisfies TraceDiagnosticsErrorSummary)
453 : undefined;
454
455 if (!foundFile) {
456 return makeEmptyDiagnostics({
457 traceFilePath: options.traceFilePath,
458 scannedFilePaths: paths,
459 readAt,
460 slowSpanThresholdMs,
461 error:
462 readFailureError ??
463 ({
464 kind: "trace-file-not-found",
465 message: "No local trace files were found.",
466 } satisfies TraceDiagnosticsErrorSummary),
467 });
468 }
469
470 return aggregator.finish({
471 traceFilePath: options.traceFilePath,
472 scannedFilePaths: paths,
473 readAt,
474 ...(readFailureError ? { partialFailure: true, error: readFailureError } : {}),
475 });
476 },
477 );
478
479 return TraceDiagnostics.of({ read });
480});
481
482export const layer = Layer.effect(TraceDiagnostics, make);
483
484export function readTraceDiagnostics(
485 options: TraceDiagnosticsOptions,
486): Effect.Effect<ServerTraceDiagnosticsResult, never, TraceDiagnostics> {
487 return Effect.gen(function* () {
488 const diagnostics = yield* TraceDiagnostics;
489 return yield* diagnostics.read(options);
490 });
491}