authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-26 01:34:28-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-26 01:34:28-07:00
log8b873eab0d74b731bc0f387be84a8561eace789f
treef3ccb88561a7de0fc6a707cb741f46e1f475395a
parent99efaeab58297c7e346ee91a4f28044faed03c59
signature Signed by PGP key B5690EEEBB952194

perf(server): opening Diagnostics no longer loads the whole trace ring into memory (#13763)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

3 files changed, 413 insertions(+), 369 deletions(-)

apps/server/src/cli/trace.ts+2-21
...@@ -13,11 +13,10 @@ import * as Effect from "effect/Effect";...@@ -13,11 +13,10 @@ import * as Effect from "effect/Effect";
13import * as FileSystem from "effect/FileSystem";13import * as FileSystem from "effect/FileSystem";
14import * as Option from "effect/Option";14import * as Option from "effect/Option";
15import * as Schema from "effect/Schema";15import * as Schema from "effect/Schema";
16import * as Stream from "effect/Stream";
17import { Command, Flag } from "effect/unstable/cli";16import { Command, Flag } from "effect/unstable/cli";
1817
19import * as ServerConfig from "../config.ts";18import * as ServerConfig from "../config.ts";
20import { toRotatedTracePaths, TraceFileReadError } from "../diagnostics/TraceDiagnostics.ts";19import { streamTraceFileLines, toRotatedTracePaths } from "../diagnostics/TraceDiagnostics.ts";
21import { resolveBaseDir } from "../os-jank.ts";20import { resolveBaseDir } from "../os-jank.ts";
22import { baseDirFlag, DurationFromString, traceFileConfig, traceMaxFilesConfig } from "./config.ts";21import { baseDirFlag, DurationFromString, traceFileConfig, traceMaxFilesConfig } from "./config.ts";
2322
...@@ -187,27 +186,9 @@ const traceSummaryCommand = Command.make("summary", {...@@ -187,27 +186,9 @@ const traceSummaryCommand = Command.make("summary", {
187 ? (yield* Clock.currentTimeMillis) - Duration.toMillis(flags.since.value)186 ? (yield* Clock.currentTimeMillis) - Duration.toMillis(flags.since.value)
188 : undefined;187 : undefined;
189 const summarizer = makeTraceSpanSummary(sinceMs);188 const summarizer = makeTraceSpanSummary(sinceMs);
190 // Stream each file so only one chunk of text is in memory at a time.
191 yield* Effect.forEach(189 yield* Effect.forEach(
192 toRotatedTracePaths(traceFilePath, yield* traceMaxFilesConfig),190 toRotatedTracePaths(traceFilePath, yield* traceMaxFilesConfig),
193 (path) =>191 (path) => streamTraceFileLines(fs, path, summarizer.addLine),
194 fs.stream(path).pipe(
195 Stream.decodeText,
196 Stream.splitLines,
197 Stream.runForEachArray((lines) => Effect.sync(() => lines.forEach(summarizer.addLine))),
198 Effect.catchTags({
199 PlatformError: (cause) =>
200 cause.reason._tag === "NotFound"
201 ? Effect.void
202 : Effect.fail(
203 new TraceFileReadError({
204 traceFilePath: path,
205 causeTag: cause.reason._tag,
206 cause,
207 }),
208 ),
209 }),
210 ),
211 { discard: true },192 { discard: true },
212 );193 );
213 const summary = summarizer.finish();194 const summary = summarizer.finish();
apps/server/src/diagnostics/TraceDiagnostics.test.ts+207-137
...@@ -7,6 +7,7 @@ import * as Logger from "effect/Logger";...@@ -7,6 +7,7 @@ import * as Logger from "effect/Logger";
7import * as Option from "effect/Option";7import * as Option from "effect/Option";
8import * as PlatformError from "effect/PlatformError";8import * as PlatformError from "effect/PlatformError";
9import * as References from "effect/References";9import * as References from "effect/References";
10import * as Stream from "effect/Stream";
1011
11import * as TraceDiagnostics from "./TraceDiagnostics.ts";12import * as TraceDiagnostics from "./TraceDiagnostics.ts";
1213
...@@ -40,72 +41,78 @@ function record(input: {...@@ -40,72 +41,78 @@ function record(input: {
40 });41 });
41}42}
4243
44const traceFilePath = "/tmp/server.trace.ndjson";
45const readAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
46
47/** Aggregates whole lines in memory, the way diagnostics read traces before streaming. */
48function aggregateLines(lines: ReadonlyArray<string>) {
49 const aggregator = TraceDiagnostics.makeTraceDiagnosticsAggregator();
50 lines.forEach(aggregator.addLine);
51 return aggregator.finish({
52 traceFilePath,
53 scannedFilePaths: TraceDiagnostics.toRotatedTracePaths(traceFilePath, 1),
54 readAt,
55 });
56}
57
58/** Reads the trace file and one rotated backup through a fake file system. */
59function readTraces(fileSystem: Partial<FileSystem.FileSystem>) {
60 return TraceDiagnostics.readTraceDiagnostics({ traceFilePath, maxFiles: 1, readAt }).pipe(
61 Effect.provide(TraceDiagnostics.layer.pipe(Layer.provide(FileSystem.layerNoop(fileSystem)))),
62 );
63}
64
43describe("TraceDiagnostics", () => {65describe("TraceDiagnostics", () => {
44 it.effect("aggregates failures, slow spans, log levels, and parse errors", () =>66 it.effect("aggregates failures, slow spans, log levels, and parse errors", () =>
45 Effect.sync(() => {67 Effect.sync(() => {
46 const diagnostics = TraceDiagnostics.aggregateTraceDiagnostics({68 const diagnostics = aggregateLines([
47 traceFilePath: "/tmp/server.trace.ndjson",69 record({
48 readAt: DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"),70 name: "server.getConfig",
49 slowSpanThresholdMs: 1_000,71 traceId: "trace-a",
50 files: [72 spanId: "span-a",
51 {73 startMs: 1_000,
52 path: "/tmp/server.trace.ndjson.1",74 durationMs: 50,
53 text: [75 }),
54 record({76 "not-json",
55 name: "server.getConfig",77 record({
56 traceId: "trace-a",78 name: "orchestration.dispatch",
57 spanId: "span-a",79 traceId: "trace-b",
58 startMs: 1_000,80 spanId: "span-b",
59 durationMs: 50,81 startMs: 2_000,
60 }),82 durationMs: 1_500,
61 "not-json",83 exit: { _tag: "Failure", cause: "Provider crashed" },
62 ].join("\n"),84 events: [
63 },85 {
64 {86 name: "provider failed",
65 path: "/tmp/server.trace.ndjson",87 timeUnixNano: ns(3_400),
66 text: [88 attributes: { "effect.logLevel": "Error" },
67 record({89 },
68 name: "orchestration.dispatch",90 ],
69 traceId: "trace-b",91 }),
70 spanId: "span-b",92 record({
71 startMs: 2_000,93 name: "orchestration.dispatch",
72 durationMs: 1_500,94 traceId: "trace-c",
73 exit: { _tag: "Failure", cause: "Provider crashed" },95 spanId: "span-c",
74 events: [96 startMs: 4_000,
75 {97 durationMs: 250,
76 name: "provider failed",98 exit: { _tag: "Failure", cause: "Provider crashed" },
77 timeUnixNano: ns(3_400),99 }),
78 attributes: { "effect.logLevel": "Error" },100 record({
79 },101 name: "git.status",
80 ],102 traceId: "trace-d",
81 }),103 spanId: "span-d",
82 record({104 startMs: 5_000,
83 name: "orchestration.dispatch",105 durationMs: 25,
84 traceId: "trace-c",106 exit: { _tag: "Interrupted", cause: "Interrupted" },
85 spanId: "span-c",107 events: [
86 startMs: 4_000,108 {
87 durationMs: 250,109 name: "status delayed",
88 exit: { _tag: "Failure", cause: "Provider crashed" },110 timeUnixNano: ns(5_010),
89 }),111 attributes: { "effect.logLevel": "Warning" },
90 record({112 },
91 name: "git.status",113 ],
92 traceId: "trace-d",114 }),
93 spanId: "span-d",115 ]);
94 startMs: 5_000,
95 durationMs: 25,
96 exit: { _tag: "Interrupted", cause: "Interrupted" },
97 events: [
98 {
99 name: "status delayed",
100 timeUnixNano: ns(5_010),
101 attributes: { "effect.logLevel": "Warning" },
102 },
103 ],
104 }),
105 ].join("\n"),
106 },
107 ],
108 });
109116
110 assert.equal(diagnostics.recordCount, 4);117 assert.equal(diagnostics.recordCount, 4);
111 assert.equal(DateTime.formatIso(diagnostics.readAt), "2026-05-05T10:00:00.000Z");118 assert.equal(DateTime.formatIso(diagnostics.readAt), "2026-05-05T10:00:00.000Z");
...@@ -139,47 +146,110 @@ describe("TraceDiagnostics", () => {...@@ -139,47 +146,110 @@ describe("TraceDiagnostics", () => {
139 );146 );
140147
141 it.effect("returns a not-found diagnostic when no files are available", () =>148 it.effect("returns a not-found diagnostic when no files are available", () =>
142 Effect.sync(() => {149 Effect.gen(function* () {
143 const diagnostics = TraceDiagnostics.aggregateTraceDiagnostics({150 const diagnostics = yield* readTraces({});
144 traceFilePath: "/tmp/missing.trace.ndjson",
145 readAt: DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"),
146 files: [],
147 });
148151
149 assert.equal(diagnostics.recordCount, 0);152 assert.equal(diagnostics.recordCount, 0);
150 assert.equal(Option.getOrUndefined(diagnostics.error)?.kind, "trace-file-not-found");153 assert.equal(Option.getOrUndefined(diagnostics.error)?.kind, "trace-file-not-found");
151 }),154 }),
152 );155 );
153156
154 it.effect("preserves full failure causes and log messages", () =>157 it.effect("streams rotated files into the same result as reading them whole", () =>
155 Effect.sync(() => {158 Effect.gen(function* () {
156 const longCause = `VcsProcessSpawnError: ${"missing executable ".repeat(80)}`.trim();159 // CRLF and LF endings plus multi-byte text, served one byte per chunk so
157 const longMessage = `provider warning: ${"retrying command ".repeat(80)}`.trim();160 // chunks split lines, line endings, and characters.
158 const diagnostics = TraceDiagnostics.aggregateTraceDiagnostics({161 const files = new Map([
159 traceFilePath: "/tmp/server.trace.ndjson",162 [
160 readAt: DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"),163 `${traceFilePath}.1`,
161 files: [164 [
162 {165 record({
163 path: "/tmp/server.trace.ndjson",166 name: "server.getConfig",
164 text: record({167 traceId: "trace-a",
165 name: "VcsProcess.run",168 spanId: "span-a",
166 traceId: "trace-long",
167 spanId: "span-long",
168 startMs: 1_000,169 startMs: 1_000,
170 durationMs: 50,
171 }),
172 "not-json",
173 record({
174 name: "orchestration.dispatch",
175 traceId: "trace-b",
176 spanId: "span-b",
177 startMs: 2_000,
178 durationMs: 1_500,
179 exit: { _tag: "Failure", cause: "Provider crashed: café 🔥" },
180 }),
181 "",
182 ].join("\r\n"),
183 ],
184 [
185 traceFilePath,
186 [
187 record({
188 name: "git.status",
189 traceId: "trace-c",
190 spanId: "span-c",
191 startMs: 3_000,
169 durationMs: 25,192 durationMs: 25,
170 exit: { _tag: "Failure", cause: longCause },193 exit: { _tag: "Interrupted", cause: "Interrupted" },
171 events: [194 events: [
172 {195 {
173 name: longMessage,196 name: "status delayed ⏳",
174 timeUnixNano: ns(1_010),197 timeUnixNano: ns(3_010),
175 attributes: { "effect.logLevel": "Warning" },198 attributes: { "effect.logLevel": "Warning" },
176 },199 },
177 ],200 ],
178 }),201 }),
179 },202 "",
203 record({
204 name: "orchestration.dispatch",
205 traceId: "trace-d",
206 spanId: "span-d",
207 startMs: 4_000,
208 durationMs: 250,
209 exit: { _tag: "Failure", cause: "Provider crashed: café 🔥" },
210 }),
211 ].join("\n"),
180 ],212 ],
213 ]);
214 const encoder = new TextEncoder();
215
216 const diagnostics = yield* readTraces({
217 stream: (path) =>
218 Stream.fromIterable(
219 Array.from(encoder.encode(files.get(path)), (byte) => Uint8Array.of(byte)),
220 ),
181 });221 });
182222
223 assert.equal(diagnostics.recordCount, 4);
224 assert.deepStrictEqual(
225 diagnostics,
226 aggregateLines([...files.values()].flatMap((text) => text.split(/\r?\n/))),
227 );
228 }),
229 );
230
231 it.effect("preserves full failure causes and log messages", () =>
232 Effect.sync(() => {
233 const longCause = `VcsProcessSpawnError: ${"missing executable ".repeat(80)}`.trim();
234 const longMessage = `provider warning: ${"retrying command ".repeat(80)}`.trim();
235 const diagnostics = aggregateLines([
236 record({
237 name: "VcsProcess.run",
238 traceId: "trace-long",
239 spanId: "span-long",
240 startMs: 1_000,
241 durationMs: 25,
242 exit: { _tag: "Failure", cause: longCause },
243 events: [
244 {
245 name: longMessage,
246 timeUnixNano: ns(1_010),
247 attributes: { "effect.logLevel": "Warning" },
248 },
249 ],
250 }),
251 ]);
252
183 assert.equal(diagnostics.latestFailures[0]?.cause, longCause);253 assert.equal(diagnostics.latestFailures[0]?.cause, longCause);
184 assert.equal(diagnostics.commonFailures[0]?.cause, longCause);254 assert.equal(diagnostics.commonFailures[0]?.cause, longCause);
185 assert.equal(diagnostics.latestWarningAndErrorLogs[0]?.message, longMessage);255 assert.equal(diagnostics.latestWarningAndErrorLogs[0]?.message, longMessage);
...@@ -188,45 +258,34 @@ describe("TraceDiagnostics", () => {...@@ -188,45 +258,34 @@ describe("TraceDiagnostics", () => {
188258
189 it.effect("keeps loaded trace data when one rotated trace file fails to read", () =>259 it.effect("keeps loaded trace data when one rotated trace file fails to read", () =>
190 Effect.gen(function* () {260 Effect.gen(function* () {
191 const traceFilePath = "/tmp/server.trace.ndjson";
192 const readFailure = PlatformError.systemError({261 const readFailure = PlatformError.systemError({
193 _tag: "PermissionDenied",262 _tag: "PermissionDenied",
194 module: "FileSystem",263 module: "FileSystem",
195 method: "readFileString",264 method: "open",
196 description: "permission denied",265 description: "permission denied",
197 pathOrDescriptor: `${traceFilePath}.1`,266 pathOrDescriptor: `${traceFilePath}.1`,
198 });267 });
199 const fileSystemLayer = FileSystem.layerNoop({
200 readFileString: (path) =>
201 path === `${traceFilePath}.1`
202 ? Effect.fail(readFailure)
203 : Effect.succeed(
204 record({
205 name: "server.getConfig",
206 traceId: "trace-a",
207 spanId: "span-a",
208 startMs: 1_000,
209 durationMs: 50,
210 }),
211 ),
212 });
213 const logAnnotations: Array<Record<string, unknown>> = [];268 const logAnnotations: Array<Record<string, unknown>> = [];
214 const logger = Logger.make<unknown, void>((options) => {269 const logger = Logger.make<unknown, void>((options) => {
215 logAnnotations.push({ ...options.fiber.getRef(References.CurrentLogAnnotations) });270 logAnnotations.push({ ...options.fiber.getRef(References.CurrentLogAnnotations) });
216 });271 });
217272
218 const diagnostics = yield* TraceDiagnostics.readTraceDiagnostics({273 const diagnostics = yield* readTraces({
219 traceFilePath,274 stream: (path) =>
220 maxFiles: 1,275 path === `${traceFilePath}.1`
221 readAt: DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"),276 ? Stream.fail(readFailure)
222 }).pipe(277 : Stream.make(
223 Effect.provide(278 new TextEncoder().encode(
224 Layer.mergeAll(279 record({
225 TraceDiagnostics.layer.pipe(Layer.provide(fileSystemLayer)),280 name: "server.getConfig",
226 Logger.layer([logger], { mergeWithExisting: false }),281 traceId: "trace-a",
227 ),282 spanId: "span-a",
228 ),283 startMs: 1_000,
229 );284 durationMs: 50,
285 }),
286 ),
287 ),
288 }).pipe(Effect.provide(Logger.layer([logger], { mergeWithExisting: false })));
230289
231 assert.equal(diagnostics.recordCount, 1);290 assert.equal(diagnostics.recordCount, 1);
232 assert.equal(291 assert.equal(
...@@ -251,32 +310,43 @@ describe("TraceDiagnostics", () => {...@@ -251,32 +310,43 @@ describe("TraceDiagnostics", () => {
251 }),310 }),
252 );311 );
253312
254 it.effect("keeps only the slowest span occurrences while aggregating large inputs", () =>313 it.effect("keeps only the top spans, failures, and warning logs from large inputs", () =>
255 Effect.sync(() => {314 Effect.sync(() => {
256 const diagnostics = TraceDiagnostics.aggregateTraceDiagnostics({315 // Shuffled, so some older records arrive after the lists are full.
257 traceFilePath: "/tmp/server.trace.ndjson",316 const indexes = Array.from({ length: 30 }, (_, step) => (step * 7) % 30);
258 readAt: DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"),317 const diagnostics = aggregateLines(
259 files: [318 indexes.map((index) =>
260 {319 record({
261 path: "/tmp/server.trace.ndjson",320 name: `span-${index}`,
262 text: Array.from({ length: 25 }, (_, index) =>321 traceId: `trace-${index}`,
263 record({322 spanId: `span-${index}`,
264 name: `span-${index}`,323 startMs: index * 1_000,
265 traceId: `trace-${index}`,324 durationMs: index,
266 spanId: `span-${index}`,325 exit: { _tag: "Failure", cause: "Provider crashed" },
267 startMs: index * 1_000,326 events: [
268 durationMs: index,327 {
269 }),328 name: `warning ${index}`,
270 ).join("\n"),329 timeUnixNano: ns(index * 1_000),
271 },330 attributes: { "effect.logLevel": "Warning" },
272 ],331 },
273 });332 ],
333 }),
334 ),
335 );
336 const newestTwenty = Array.from({ length: 20 }, (_, rank) => `trace-${29 - rank}`);
274337
275 assert.equal(diagnostics.recordCount, 25);338 assert.equal(diagnostics.recordCount, 30);
276 assert.equal(diagnostics.slowestSpans.length, 10);
277 assert.deepStrictEqual(339 assert.deepStrictEqual(
278 diagnostics.slowestSpans.map((span) => span.durationMs),340 diagnostics.slowestSpans.map((span) => span.durationMs),
279 [24, 23, 22, 21, 20, 19, 18, 17, 16, 15],341 [29, 28, 27, 26, 25, 24, 23, 22, 21, 20],
342 );
343 assert.deepStrictEqual(
344 diagnostics.latestFailures.map((failure) => failure.traceId),
345 newestTwenty,
346 );
347 assert.deepStrictEqual(
348 diagnostics.latestWarningAndErrorLogs.map((log) => log.traceId),
349 newestTwenty,
280 );350 );
281 }),351 }),
282 );352 );
apps/server/src/diagnostics/TraceDiagnostics.ts+204-211
...@@ -16,6 +16,7 @@ import * as Option from "effect/Option";...@@ -16,6 +16,7 @@ import * as Option from "effect/Option";
16import * as PlatformError from "effect/PlatformError";16import * as PlatformError from "effect/PlatformError";
17import * as Result from "effect/Result";17import * as Result from "effect/Result";
18import * as Schema from "effect/Schema";18import * as Schema from "effect/Schema";
19import * as Stream from "effect/Stream";
1920
20interface TraceRecordLike {21interface TraceRecordLike {
21 readonly name?: unknown;22 readonly name?: unknown;
...@@ -63,16 +64,6 @@ export class TraceDiagnostics extends Context.Service<...@@ -63,16 +64,6 @@ export class TraceDiagnostics extends Context.Service<
63 }64 }
64>()("t3/diagnostics/TraceDiagnostics") {}65>()("t3/diagnostics/TraceDiagnostics") {}
6566
66interface TraceDiagnosticsInput {
67 readonly traceFilePath: string;
68 readonly files: ReadonlyArray<{ readonly path: string; readonly text: string }>;
69 readonly scannedFilePaths?: ReadonlyArray<string>;
70 readonly slowSpanThresholdMs?: number;
71 readonly readAt: DateTime.Utc;
72 readonly error?: TraceDiagnosticsErrorSummary;
73 readonly partialFailure?: boolean;
74}
75
76interface TraceDiagnosticsErrorSummary {67interface TraceDiagnosticsErrorSummary {
77 readonly kind: ServerTraceDiagnosticsErrorKind;68 readonly kind: ServerTraceDiagnosticsErrorKind;
78 readonly message: string;69 readonly message: string;
...@@ -173,44 +164,50 @@ function isNotFoundError(error: PlatformError.PlatformError): boolean {...@@ -173,44 +164,50 @@ function isNotFoundError(error: PlatformError.PlatformError): boolean {
173 return error.reason._tag === "NotFound";164 return error.reason._tag === "NotFound";
174}165}
175166
176function insertBoundedSlowestSpan(167/**
177 slowestSpans: ServerTraceDiagnosticsSpanOccurrence[],168 * Adds `item` to `items`, which stays sorted by `order` and holds at most
178 span: ServerTraceDiagnosticsSpanOccurrence,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,
179): void {177): void {
180 if (178 if (items.length >= limit && order(item, items[items.length - 1]!) >= 0) {
181 slowestSpans.length >= TOP_LIMIT &&
182 span.durationMs <= slowestSpans[slowestSpans.length - 1]!.durationMs
183 ) {
184 return;179 return;
185 }180 }
186181
187 slowestSpans.push(span);182 items.push(item);
188 slowestSpans.sort((left, right) => right.durationMs - left.durationMs);183 items.sort(order);
189 if (slowestSpans.length > TOP_LIMIT) {184 if (items.length > limit) {
190 slowestSpans.length = TOP_LIMIT;185 items.length = limit;
191 }186 }
192}187}
193188
194export function aggregateTraceDiagnostics(189const slowestFirst = (
195 input: TraceDiagnosticsInput,190 left: ServerTraceDiagnosticsSpanOccurrence,
196): ServerTraceDiagnosticsResult {191 right: ServerTraceDiagnosticsSpanOccurrence,
197 const readAt = input.readAt;192) => right.durationMs - left.durationMs;
198 const slowSpanThresholdMs = input.slowSpanThresholdMs ?? DEFAULT_SLOW_SPAN_THRESHOLD_MS;193
199 const scannedFilePaths = input.scannedFilePaths ?? input.files.map((file) => file.path);194const latestEndedFirst = (
200 if (input.files.length === 0) {195 left: ServerTraceDiagnosticsRecentFailure,
201 return makeEmptyDiagnostics({196 right: ServerTraceDiagnosticsRecentFailure,
202 traceFilePath: input.traceFilePath,197) => DateTime.toEpochMillis(right.endedAt) - DateTime.toEpochMillis(left.endedAt);
203 scannedFilePaths,198
204 readAt,199const latestSeenFirst = (
205 slowSpanThresholdMs,200 left: ServerTraceDiagnosticsLogEvent,
206 error: input.error ?? {201 right: ServerTraceDiagnosticsLogEvent,
207 kind: "trace-file-not-found",202) => DateTime.toEpochMillis(right.seenAt) - DateTime.toEpochMillis(left.seenAt);
208 message: "No local trace files were found.",203
209 },204/**
210 ...(input.partialFailure ? { partialFailure: true } : {}),205 * Folds trace NDJSON into diagnostics. Call `addLine` once per line as the
211 });206 * rotated files stream in, then `finish` for the result.
212 }207 */
213208export function makeTraceDiagnosticsAggregator(
209 slowSpanThresholdMs = DEFAULT_SLOW_SPAN_THRESHOLD_MS,
210) {
214 let parseErrorCount = 0;211 let parseErrorCount = 0;
215 let recordCount = 0;212 let recordCount = 0;
216 let failureCount = 0;213 let failureCount = 0;
...@@ -229,182 +226,184 @@ export function aggregateTraceDiagnostics(...@@ -229,182 +226,184 @@ export function aggregateTraceDiagnostics(
229 const latestWarningAndErrorLogs: ServerTraceDiagnosticsLogEvent[] = [];226 const latestWarningAndErrorLogs: ServerTraceDiagnosticsLogEvent[] = [];
230 const logLevelCounts: Record<string, number> = {};227 const logLevelCounts: Record<string, number> = {};
231228
232 for (const file of input.files) {229 const addLine = (line: string) => {
233 const lines = file.text.split(/\r?\n/);230 if (line.trim().length === 0) return;
234 for (const line of lines) {
235 if (line.trim().length === 0) continue;
236
237 let parsed: unknown;
238 try {
239 parsed = JSON.parse(line);
240 } catch {
241 parseErrorCount += 1;
242 continue;
243 }
244231
245 if (!isRecordObject(parsed)) {232 let parsed: unknown;
246 parseErrorCount += 1;233 try {
247 continue;234 parsed = JSON.parse(line);
248 }235 } catch {
236 parseErrorCount += 1;
237 return;
238 }
249239
250 const name = toStringValue(parsed.name);240 if (!isRecordObject(parsed)) {
251 const traceId = toStringValue(parsed.traceId);241 parseErrorCount += 1;
252 const spanId = toStringValue(parsed.spanId);242 return;
253 const durationMs = toNumberValue(parsed.durationMs);243 }
254 const endedAt = unixNanoToDateTime(parsed.endTimeUnixNano);
255 const startedAt = unixNanoToDateTime(parsed.startTimeUnixNano);
256244
257 if (!name || !traceId || !spanId || durationMs === null || !endedAt) {245 const name = toStringValue(parsed.name);
258 parseErrorCount += 1;246 const traceId = toStringValue(parsed.traceId);
259 continue;247 const spanId = toStringValue(parsed.spanId);
260 }248 const durationMs = toNumberValue(parsed.durationMs);
249 const endedAt = unixNanoToDateTime(parsed.endTimeUnixNano);
250 const startedAt = unixNanoToDateTime(parsed.startTimeUnixNano);
261251
262 recordCount += 1;252 if (!name || !traceId || !spanId || durationMs === null || !endedAt) {
263 firstSpanAt =253 parseErrorCount += 1;
264 startedAt && (firstSpanAt === null || DateTime.isLessThan(startedAt, firstSpanAt))254 return;
265 ? startedAt255 }
266 : firstSpanAt;
267 lastSpanAt =
268 lastSpanAt === null || DateTime.isGreaterThan(endedAt, lastSpanAt) ? endedAt : lastSpanAt;
269
270 const exitTag = readExitTag(parsed.exit);
271 const isFailure = exitTag === "Failure";
272 const isInterrupted = exitTag === "Interrupted";
273 if (isFailure) failureCount += 1;
274 if (isInterrupted) interruptionCount += 1;
275
276 const spanSummary = spansByName.get(name) ?? {
277 count: 0,
278 failureCount: 0,
279 totalDurationMs: 0,
280 maxDurationMs: 0,
281 };
282 spanSummary.count += 1;
283 spanSummary.totalDurationMs += durationMs;
284 spanSummary.maxDurationMs = Math.max(spanSummary.maxDurationMs, durationMs);
285 if (isFailure) spanSummary.failureCount += 1;
286 spansByName.set(name, spanSummary);
287
288 const spanItem = { name, durationMs, endedAt, traceId, spanId };
289 if (durationMs >= slowSpanThresholdMs) {
290 slowSpanCount += 1;
291 }
292 insertBoundedSlowestSpan(slowestSpans, spanItem);
293
294 if (isFailure) {
295 const cause = readExitCause(parsed.exit);
296 latestFailures.push({ ...spanItem, cause });
297
298 const failureKey = `${name}\0${cause}`;
299 const existing = failuresByKey.get(failureKey);
300 const isLatestFailure = !existing || DateTime.isGreaterThan(endedAt, existing.lastSeenAt);
301 failuresByKey.set(failureKey, {
302 name,
303 cause,
304 count: (existing?.count ?? 0) + 1,
305 lastSeenAt: isLatestFailure ? endedAt : existing!.lastSeenAt,
306 traceId: isLatestFailure ? traceId : existing!.traceId,
307 spanId: isLatestFailure ? spanId : existing!.spanId,
308 });
309 }
310256
311 if (Array.isArray(parsed.events)) {257 recordCount += 1;
312 for (const rawEvent of parsed.events) {258 firstSpanAt =
313 if (!isTraceEvent(rawEvent)) continue;259 startedAt && (firstSpanAt === null || DateTime.isLessThan(startedAt, firstSpanAt))
314 const attributes = readEventAttributes(rawEvent);260 ? startedAt
315 const level = toStringValue(attributes["effect.logLevel"]);261 : firstSpanAt;
316 if (!level) continue;262 lastSpanAt =
317263 lastSpanAt === null || DateTime.isGreaterThan(endedAt, lastSpanAt) ? endedAt : lastSpanAt;
318 logLevelCounts[level] = (logLevelCounts[level] ?? 0) + 1;264
319 const normalizedLevel = level.toLowerCase();265 const exitTag = readExitTag(parsed.exit);
320 if (266 const isFailure = exitTag === "Failure";
321 normalizedLevel !== "warning" &&267 const isInterrupted = exitTag === "Interrupted";
322 normalizedLevel !== "warn" &&268 if (isFailure) failureCount += 1;
323 normalizedLevel !== "error" &&269 if (isInterrupted) interruptionCount += 1;
324 normalizedLevel !== "fatal"270
325 ) {271 const spanSummary = spansByName.get(name) ?? {
326 continue;272 count: 0,
327 }273 failureCount: 0,
328274 totalDurationMs: 0,
329 const seenAt = unixNanoToDateTime(rawEvent.timeUnixNano) ?? endedAt;275 maxDurationMs: 0,
330 const message = toStringValue(rawEvent.name)?.trim() ?? "Log event";276 };
331 latestWarningAndErrorLogs.push({277 spanSummary.count += 1;
332 spanName: name,278 spanSummary.totalDurationMs += durationMs;
333 level,279 spanSummary.maxDurationMs = Math.max(spanSummary.maxDurationMs, durationMs);
334 message,280 if (isFailure) spanSummary.failureCount += 1;
335 seenAt,281 spansByName.set(name, spanSummary);
336 traceId,282
337 spanId,283 const spanItem = { name, durationMs, endedAt, traceId, spanId };
338 });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;
339 }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 );
340 }332 }
341 }333 }
342 }334 };
343
344 const topSpansByCount: ServerTraceDiagnosticsSpanSummary[] = [...spansByName.entries()]
345 .map(([name, span]) => ({
346 name,
347 count: span.count,
348 failureCount: span.failureCount,
349 totalDurationMs: span.totalDurationMs,
350 averageDurationMs: span.count > 0 ? span.totalDurationMs / span.count : 0,
351 maxDurationMs: span.maxDurationMs,
352 }))
353 .toSorted((left, right) => right.count - left.count || right.maxDurationMs - left.maxDurationMs)
354 .slice(0, TOP_LIMIT);
355335
356 return {336 const finish = (input: {
357 traceFilePath: input.traceFilePath,337 readonly traceFilePath: string;
358 scannedFilePaths,338 readonly scannedFilePaths: ReadonlyArray<string>;
359 readAt,339 readonly readAt: DateTime.Utc;
360 recordCount,340 readonly error?: TraceDiagnosticsErrorSummary;
361 parseErrorCount,341 readonly partialFailure?: boolean;
362 firstSpanAt: Option.fromNullishOr(firstSpanAt),342 }): ServerTraceDiagnosticsResult => {
363 lastSpanAt: Option.fromNullishOr(lastSpanAt),343 const topSpansByCount: ServerTraceDiagnosticsSpanSummary[] = [...spansByName.entries()]
364 failureCount,344 .map(([name, span]) => ({
365 interruptionCount,345 name,
366 slowSpanThresholdMs,346 count: span.count,
367 slowSpanCount,347 failureCount: span.failureCount,
368 logLevelCounts,348 totalDurationMs: span.totalDurationMs,
369 topSpansByCount,349 averageDurationMs: span.count > 0 ? span.totalDurationMs / span.count : 0,
370 slowestSpans,350 maxDurationMs: span.maxDurationMs,
371 commonFailures: [...failuresByKey.values()]351 }))
372 .toSorted(
373 (left, right) =>
374 right.count - left.count ||
375 DateTime.toEpochMillis(right.lastSeenAt) - DateTime.toEpochMillis(left.lastSeenAt),
376 )
377 .slice(0, TOP_LIMIT),
378 latestFailures: latestFailures
379 .toSorted(352 .toSorted(
380 (left, right) =>353 (left, right) => right.count - left.count || right.maxDurationMs - left.maxDurationMs,
381 DateTime.toEpochMillis(right.endedAt) - DateTime.toEpochMillis(left.endedAt),
382 )354 )
383 .slice(0, RECENT_LIMIT),355 .slice(0, TOP_LIMIT);
384 latestWarningAndErrorLogs: latestWarningAndErrorLogs356
385 .toSorted(357 return {
386 (left, right) => DateTime.toEpochMillis(right.seenAt) - DateTime.toEpochMillis(left.seenAt),358 traceFilePath: input.traceFilePath,
387 )359 scannedFilePaths: input.scannedFilePaths,
388 .slice(0, RECENT_LIMIT),360 readAt: input.readAt,
389 partialFailure: input.partialFailure ? Option.some(true) : Option.none(),361 recordCount,
390 error: Option.fromNullishOr(input.error),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 };
391 };384 };
392}
393385
394type TraceFileReadResult =386 return { addLine, finish };
395 | { readonly _tag: "Loaded"; readonly path: string; readonly text: string }387}
396 | { readonly _tag: "Missing"; readonly path: string };
397388
398function readTraceFile(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(
399 fileSystem: FileSystem.FileSystem,394 fileSystem: FileSystem.FileSystem,
400 path: string,395 path: string,
401): Effect.Effect<TraceFileReadResult, TraceFileReadError> {396 onLine: (line: string) => void,
402 return fileSystem.readFileString(path).pipe(397): Effect.Effect<boolean, TraceFileReadError> {
403 Effect.map((text): TraceFileReadResult => ({ _tag: "Loaded", path, text })),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),
404 Effect.catchTags({403 Effect.catchTags({
405 PlatformError: (cause) =>404 PlatformError: (cause) =>
406 isNotFoundError(cause)405 isNotFoundError(cause)
407 ? Effect.succeed<TraceFileReadResult>({ _tag: "Missing", path })406 ? Effect.succeed(false)
408 : Effect.fail(407 : Effect.fail(
409 new TraceFileReadError({408 new TraceFileReadError({
410 traceFilePath: path,409 traceFilePath: path,
...@@ -425,10 +424,11 @@ export const make = Effect.gen(function* () {...@@ -425,10 +424,11 @@ export const make = Effect.gen(function* () {
425 const readAt = options.readAt ?? (yield* DateTime.now);424 const readAt = options.readAt ?? (yield* DateTime.now);
426 const slowSpanThresholdMs = options.slowSpanThresholdMs ?? DEFAULT_SLOW_SPAN_THRESHOLD_MS;425 const slowSpanThresholdMs = options.slowSpanThresholdMs ?? DEFAULT_SLOW_SPAN_THRESHOLD_MS;
427 const paths = toRotatedTracePaths(options.traceFilePath, options.maxFiles);426 const paths = toRotatedTracePaths(options.traceFilePath, options.maxFiles);
427 const aggregator = makeTraceDiagnosticsAggregator(slowSpanThresholdMs);
428 const results = yield* Effect.forEach(428 const results = yield* Effect.forEach(
429 paths,429 paths,
430 (path) =>430 (path) =>
431 readTraceFile(fileSystem, path).pipe(431 streamTraceFileLines(fileSystem, path, aggregator.addLine).pipe(
432 Effect.tapError((cause) =>432 Effect.tapError((cause) =>
433 Effect.logWarning("Failed to read local trace file.").pipe(433 Effect.logWarning("Failed to read local trace file.").pipe(
434 Effect.annotateLogs({434 Effect.annotateLogs({
...@@ -440,15 +440,10 @@ export const make = Effect.gen(function* () {...@@ -440,15 +440,10 @@ export const make = Effect.gen(function* () {
440 ),440 ),
441 Effect.result,441 Effect.result,
442 ),442 ),
443 {443 // Every file feeds one aggregator, so read them one at a time, oldest first.
444 concurrency: 1,444 { concurrency: 1 },
445 },
446 );
447 const files = results.flatMap((result) =>
448 Result.isSuccess(result) && result.success._tag === "Loaded"
449 ? [{ path: result.success.path, text: result.success.text }]
450 : [],
451 );445 );
446 const foundFile = results.some((result) => Result.isSuccess(result) && result.success);
452 const readFailure = results.find(Result.isFailure);447 const readFailure = results.find(Result.isFailure);
453 const readFailureError = readFailure448 const readFailureError = readFailure
454 ? ({449 ? ({
...@@ -457,7 +452,7 @@ export const make = Effect.gen(function* () {...@@ -457,7 +452,7 @@ export const make = Effect.gen(function* () {
457 } satisfies TraceDiagnosticsErrorSummary)452 } satisfies TraceDiagnosticsErrorSummary)
458 : undefined;453 : undefined;
459454
460 if (files.length === 0) {455 if (!foundFile) {
461 return makeEmptyDiagnostics({456 return makeEmptyDiagnostics({
462 traceFilePath: options.traceFilePath,457 traceFilePath: options.traceFilePath,
463 scannedFilePaths: paths,458 scannedFilePaths: paths,
...@@ -472,12 +467,10 @@ export const make = Effect.gen(function* () {...@@ -472,12 +467,10 @@ export const make = Effect.gen(function* () {
472 });467 });
473 }468 }
474469
475 return aggregateTraceDiagnostics({470 return aggregator.finish({
476 traceFilePath: options.traceFilePath,471 traceFilePath: options.traceFilePath,
477 files,
478 scannedFilePaths: paths,472 scannedFilePaths: paths,
479 readAt,473 readAt,
480 slowSpanThresholdMs,
481 ...(readFailureError ? { partialFailure: true, error: readFailureError } : {}),474 ...(readFailureError ? { partialFailure: true, error: readFailureError } : {}),
482 });475 });
483 },476 },