authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 05:37:03-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 05:37:03-07:00
log4befc863f54a3341911c346c8822286ec37e54af
treee410dc10e49587f03d09bc8eb15e58f8049ba733
parent8414d168191599dec08588080d0b77506c849424
signature Signed by PGP key B5690EEEBB952194

perf(server): traces keep ~40% more history and no longer block the event loop (#19430)

Co-authored-by: Aaron Queen <bompus@users.noreply.github.com> Co-authored-by: Daniyal <115576848+daniyal-sh@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

11 files changed, 552 insertions(+), 176 deletions(-)

apps/server/src/orchestration-v2/EventSink.ts+6-1
...@@ -399,7 +399,12 @@ const layerBase: Layer.Layer<...@@ -399,7 +399,12 @@ const layerBase: Layer.Layer<
399 yield* applyStoredEvents(committed);399 yield* applyStoredEvents(committed);
400 yield* effectOutbox.enqueue(input.effects);400 yield* effectOutbox.enqueue(input.effects);
401 return committed;401 return committed;
402 }),402 }).pipe(
403 // Every streamed agent event is a write, so its statements would
404 // dominate the trace file. This span and its sql.transaction span
405 // still time each write, including the commit.
406 Effect.withTracerEnabled(false),
407 ),
403 (storedEvents) =>408 (storedEvents) =>
404 Effect.gen(function* () {409 Effect.gen(function* () {
405 if (input.effects.length > 0) {410 if (input.effects.length > 0) {
apps/server/src/orchestration-v2/FoundationPersistence.test.ts+26
...@@ -447,6 +447,32 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {...@@ -447,6 +447,32 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
447 ),447 ),
448 );448 );
449449
450 it.effect("traces an event write and its transaction without its SQL statements", () =>
451 Effect.gen(function* () {
452 const sink = yield* EventSink.EventSinkV2;
453 const now = yield* DateTime.now;
454 const thread = makeThread(ThreadId.make("thread:untraced-write-statements"), now);
455 const names: Array<string> = [];
456 const tracer = Tracer.make({
457 span: (options) => {
458 names.push(options.name);
459 return new Tracer.NativeSpan(options);
460 },
461 });
462
463 yield* sink
464 .write({
465 events: [
466 threadCreatedEvent({ id: "event:untraced-write-statements:created", thread, now }),
467 ],
468 })
469 .pipe(Effect.withTracer(tracer));
470
471 // These two spans time a slow commit; the statements inside are not traced.
472 assert.deepEqual(names, ["orchestrationV2.EventSink.write", "sql.transaction"]);
473 }),
474 );
475
450 it.effect.each(["high-water", "replay"] as const)(476 it.effect.each(["high-water", "replay"] as const)(
451 "bounds live events while the V2 %s query is blocked",477 "bounds live events while the V2 %s query is blocked",
452 (phase) =>478 (phase) =>
apps/server/src/provider/EventNdjsonLogger.test.ts+93-29
...@@ -4,9 +4,13 @@ import * as NodeOS from "node:os";...@@ -4,9 +4,13 @@ import * as NodeOS from "node:os";
4import * as NodePath from "node:path";4import * as NodePath from "node:path";
55
6import { ThreadId } from "@t3tools/contracts";6import { ThreadId } from "@t3tools/contracts";
7import { assert, describe, it } from "@effect/vitest";7import * as NodeServices from "@effect/platform-node/NodeServices";
8import { assert, it } from "@effect/vitest";
9import { RotatingFileSinkError } from "@t3tools/shared/logging";
8import * as Clock from "effect/Clock";10import * as Clock from "effect/Clock";
9import * as Effect from "effect/Effect";11import * as Effect from "effect/Effect";
12import * as Deferred from "effect/Deferred";
13import * as FileSystem from "effect/FileSystem";
10import * as Fiber from "effect/Fiber";14import * as Fiber from "effect/Fiber";
11import * as Logger from "effect/Logger";15import * as Logger from "effect/Logger";
12import * as Schema from "effect/Schema";16import * as Schema from "effect/Schema";
...@@ -49,7 +53,7 @@ function parseLogLine(line: string) {...@@ -49,7 +53,7 @@ function parseLogLine(line: string) {
49 };53 };
50}54}
5155
52describe("EventNdjsonLogger", () => {56it.layer(NodeServices.layer)("EventNdjsonLogger", (it) => {
53 it.effect("summarizes circular events without exposing their contents in diagnostics", () => {57 it.effect("summarizes circular events without exposing their contents in diagnostics", () => {
54 const messages: Array<unknown> = [];58 const messages: Array<unknown> = [];
55 const logCapture = Logger.make<unknown, void>(({ message }) => {59 const logCapture = Logger.make<unknown, void>(({ message }) => {
...@@ -256,8 +260,8 @@ describe("EventNdjsonLogger", () => {...@@ -256,8 +260,8 @@ describe("EventNdjsonLogger", () => {
256260
257 assert.equal(NodeFS.existsSync(threadPath), false);261 assert.equal(NodeFS.existsSync(threadPath), false);
258 yield* TestClock.adjust(1_000);262 yield* TestClock.adjust(1_000);
259 assert.equal(NodeFS.existsSync(threadPath), true);
260 yield* store.close();263 yield* store.close();
264 assert.equal(NodeFS.existsSync(threadPath), true);
261 } finally {265 } finally {
262 NodeFS.rmSync(tempDir, { recursive: true, force: true });266 NodeFS.rmSync(tempDir, { recursive: true, force: true });
263 }267 }
...@@ -281,10 +285,10 @@ describe("EventNdjsonLogger", () => {...@@ -281,10 +285,10 @@ describe("EventNdjsonLogger", () => {
281 yield* logger.write({ id: "accepted" }, ThreadId.make("thread-interrupted"));285 yield* logger.write({ id: "accepted" }, ThreadId.make("thread-interrupted"));
282286
283 yield* TestClock.adjust(1_000);287 yield* TestClock.adjust(1_000);
288 yield* store.close();
284289
285 assert.equal(NodeFS.existsSync(threadPath), true);290 assert.equal(NodeFS.existsSync(threadPath), true);
286 assert.include(NodeFS.readFileSync(threadPath, "utf8"), '{"id":"accepted"}');291 assert.include(NodeFS.readFileSync(threadPath, "utf8"), '{"id":"accepted"}');
287 yield* store.close();
288 } finally {292 } finally {
289 NodeFS.rmSync(tempDir, { recursive: true, force: true });293 NodeFS.rmSync(tempDir, { recursive: true, force: true });
290 }294 }
...@@ -788,39 +792,99 @@ describe("EventNdjsonLogger", () => {...@@ -788,39 +792,99 @@ describe("EventNdjsonLogger", () => {
788 }),792 }),
789 );793 );
790794
791 it("attributes batches that were written before a later chunk fails", () => {795 it.effect("attributes batches that were written before a later chunk fails", () =>
792 const records: ReadonlyArray<PendingRecord> = [796 Effect.gen(function* () {
793 {797 const records: ReadonlyArray<PendingRecord> = [
794 stream: "native",798 {
795 threadSegment: "thread",799 stream: "native",
796 line: "first",800 threadSegment: "thread",
797 bytes: 5,801 line: "first",
798 },802 bytes: 5,
799 {803 },
800 stream: "canonical",804 {
801 threadSegment: "thread",805 stream: "canonical",
802 line: "second",806 threadSegment: "thread",
803 bytes: 6,807 line: "second",
804 },808 bytes: 6,
805 ];809 },
806 const attributed: Array<PendingRecord> = [];810 ];
807 let writes = 0;811 const attributed: Array<PendingRecord> = [];
808812 let writes = 0;
809 assert.throws(() =>813
810 writeBatchedMessages(814 const result = yield* writeBatchedMessages(
811 {815 {
812 write: () => {816 write: () => {
813 writes += 1;817 writes += 1;
814 if (writes === 2) throw new Error("simulated disk exhaustion");818 return writes === 2
819 ? Effect.fail(
820 new RotatingFileSinkError({
821 operation: "write",
822 filePath: "test.log",
823 cause: "simulated disk exhaustion",
824 }),
825 )
826 : Effect.void;
815 },827 },
816 },828 },
817 records,829 records,
818 5,830 5,
819 (written) => attributed.push(...written),831 (written) => attributed.push(...written),
820 ),832 ).pipe(Effect.result);
821 );833 assert.equal(result._tag, "Failure");
822 assert.deepEqual(attributed, [records[0]]);834 assert.deepEqual(attributed, [records[0]]);
823 });835 }),
836 );
837
838 it.effect(
839 "keeps unrelated work responsive and drains before close while a disk write is held",
840 () =>
841 Effect.gen(function* () {
842 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
843 const basePath = NodePath.join(tempDir, "events.log");
844 const fs = yield* FileSystem.FileSystem;
845 const started = yield* Deferred.make<void>();
846 const release = yield* Deferred.make<void>();
847 let writes = 0;
848 try {
849 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 }).pipe(
850 Effect.provideService(FileSystem.FileSystem, {
851 ...fs,
852 writeFile: (filePath, content, options) =>
853 Effect.gen(function* () {
854 if (++writes === 1) {
855 yield* Deferred.succeed(started, undefined);
856 yield* Deferred.await(release);
857 }
858 yield* fs.writeFile(filePath, content, options);
859 }),
860 }),
861 );
862 const first = yield* store
863 .logger("native")
864 .write({ id: "first" }, ThreadId.make("held"))
865 .pipe(Effect.forkChild);
866 yield* Deferred.await(started);
867 const closed = yield* Deferred.make<void>();
868 const close = yield* store
869 .close()
870 .pipe(Effect.andThen(Deferred.succeed(closed, undefined)), Effect.forkChild);
871 assert.equal(yield* Deferred.isDone(closed), false);
872 // A separate fiber runs while the file operation remains pending.
873 const heartbeat = yield* Deferred.make<void>();
874 yield* Effect.forkChild(Deferred.succeed(heartbeat, undefined));
875 yield* Deferred.await(heartbeat);
876 yield* Deferred.succeed(release, undefined);
877 yield* Fiber.join(first);
878 yield* Fiber.join(close);
879 assert.include(
880 NodeFS.readFileSync(ownedLogPath(basePath, "held"), "utf8"),
881 '{"id":"first"}',
882 );
883 } finally {
884 NodeFS.rmSync(tempDir, { recursive: true, force: true });
885 }
886 }),
887 );
824888
825 it.effect("reports logical provider log writes to resource attribution", () =>889 it.effect("reports logical provider log writes to resource attribution", () =>
826 Effect.gen(function* () {890 Effect.gen(function* () {
apps/server/src/provider/EventNdjsonLogger.ts+106-76
...@@ -10,11 +10,14 @@ import * as NodePath from "node:path";...@@ -10,11 +10,14 @@ import * as NodePath from "node:path";
1010
11import type { ThreadId } from "@t3tools/contracts";11import type { ThreadId } from "@t3tools/contracts";
12import type * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";12import type * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
13import { RotatingFileSink } from "@t3tools/shared/logging";13import { makeAsyncRotatingFileSink } from "@t3tools/shared/logging";
14import { errorTag } from "@t3tools/shared/observability";14import { errorTag } from "@t3tools/shared/observability";
15import * as Clock from "effect/Clock";15import * as Clock from "effect/Clock";
16import * as DateTime from "effect/DateTime";16import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";17import * as Effect from "effect/Effect";
18import * as FileSystem from "effect/FileSystem";
19import * as Path from "effect/Path";
20import * as Result from "effect/Result";
18import * as Exit from "effect/Exit";21import * as Exit from "effect/Exit";
19import * as Schema from "effect/Schema";22import * as Schema from "effect/Schema";
20import * as Scope from "effect/Scope";23import * as Scope from "effect/Scope";
...@@ -135,10 +138,12 @@ export interface PendingRecord {...@@ -135,10 +138,12 @@ export interface PendingRecord {
135 readonly bytes: number;138 readonly bytes: number;
136}139}
137140
141type AsyncFileSink = Effect.Success<ReturnType<typeof makeAsyncRotatingFileSink>>;
142
138interface StoreState {143interface StoreState {
139 readonly pending: Array<PendingRecord>;144 readonly pending: Array<PendingRecord>;
140 readonly pendingBytes: number;145 readonly pendingBytes: number;
141 readonly sinks: ReadonlyMap<string, RotatingFileSink>;146 readonly sinks: ReadonlyMap<string, AsyncFileSink>;
142 readonly flushScheduled: boolean;147 readonly flushScheduled: boolean;
143 readonly closed: boolean;148 readonly closed: boolean;
144 readonly lastRetentionAt: number;149 readonly lastRetentionAt: number;
...@@ -369,65 +374,69 @@ export function boundProviderEventForLogging(event: unknown): unknown {...@@ -369,65 +374,69 @@ export function boundProviderEventForLogging(event: unknown): unknown {
369 return summarizeProviderEvent(event);374 return summarizeProviderEvent(event);
370}375}
371376
372export function writeBatchedMessages(377export const writeBatchedMessages = Effect.fnUntraced(function* (
373 sink: Pick<RotatingFileSink, "write">,378 sink: AsyncFileSink,
374 records: ReadonlyArray<PendingRecord>,379 records: ReadonlyArray<PendingRecord>,
375 maxBytes: number,380 maxBytes: number,
376 onWritten: (records: ReadonlyArray<PendingRecord>) => void,381 onWritten: (records: ReadonlyArray<PendingRecord>) => void,
377): void {382) {
378 let pendingRecords: Array<PendingRecord> = [];383 let pendingRecords: Array<PendingRecord> = [];
379 let pendingBytes = 0;384 let pendingBytes = 0;
380385
381 const flush = () => {386 const flush = Effect.gen(function* () {
382 if (pendingRecords.length === 0) return;387 if (pendingRecords.length === 0) return;
383 const writtenRecords = pendingRecords;388 const writtenRecords = pendingRecords;
384 sink.write(writtenRecords.map((record) => record.line).join(""));389 yield* sink.write(writtenRecords.map((record) => record.line).join(""));
385 onWritten(writtenRecords);390 onWritten(writtenRecords);
386 pendingRecords = [];391 pendingRecords = [];
387 pendingBytes = 0;392 pendingBytes = 0;
388 };393 });
389394
390 for (const record of records) {395 for (const record of records) {
391 if (pendingBytes > 0 && pendingBytes + record.bytes > maxBytes) {396 if (pendingBytes > 0 && pendingBytes + record.bytes > maxBytes) {
392 flush();397 yield* flush;
393 }398 }
394 pendingRecords.push(record);399 pendingRecords.push(record);
395 pendingBytes += record.bytes;400 pendingBytes += record.bytes;
396 if (pendingBytes >= maxBytes) {401 if (pendingBytes >= maxBytes) {
397 flush();402 yield* flush;
398 }403 }
399 }404 }
400 flush();405 yield* flush;
401}406});
402407
403function isProviderLogFile(filePath: string, fileName: string, filePrefix: string): boolean {408async function isProviderLogFile(
409 filePath: string,
410 fileName: string,
411 filePrefix: string,
412): Promise<boolean> {
404 if (!/\.log(?:\.\d+)?$/u.test(fileName)) return false;413 if (!/\.log(?:\.\d+)?$/u.test(fileName)) return false;
405 if (fileName.startsWith(filePrefix)) return true;414 if (fileName.startsWith(filePrefix)) return true;
406415
407 const descriptor = NodeFS.openSync(filePath, "r");416 const descriptor = await NodeFS.promises.open(filePath, "r");
408 try {417 try {
409 const header = Buffer.alloc(256);418 const header = Buffer.alloc(256);
410 const bytesRead = NodeFS.readSync(descriptor, header, 0, header.byteLength, 0);419 const { bytesRead } = await descriptor.read(header, 0, header.byteLength, 0);
411 return /^\[[^\]\r\n]+\] (?:NTIVE|CANON|ORCH): /u.test(header.toString("utf8", 0, bytesRead));420 return /^\[[^\]\r\n]+\] (?:NTIVE|CANON|ORCH): /u.test(header.toString("utf8", 0, bytesRead));
412 } finally {421 } finally {
413 NodeFS.closeSync(descriptor);422 await descriptor.close();
414 }423 }
415}424}
416425
417function enforceRetention(input: {426async function enforceRetention(input: {
418 readonly directory: string;427 readonly directory: string;
419 readonly maxTotalBytes: number;428 readonly maxTotalBytes: number;
420 readonly maxAgeMs: number;429 readonly maxAgeMs: number;
421 readonly activeFilePaths: ReadonlySet<string>;430 readonly activeFilePaths: ReadonlySet<string>;
422 readonly filePrefix: string;431 readonly filePrefix: string;
423 readonly now: number;432 readonly now: number;
424}): RetentionResult {433}): Promise<RetentionResult> {
425 const failures: Array<FileOperationFailure> = [];434 const failures: Array<FileOperationFailure> = [];
426 const files: Array<{ filePath: string; mtimeMs: number; size: number }> = [];435 const files: Array<{ filePath: string; mtimeMs: number; size: number }> = [];
427436
428 let entries: ReadonlyArray<NodeFS.Dirent>;437 let entries: ReadonlyArray<NodeFS.Dirent>;
429 try {438 try {
430 entries = NodeFS.readdirSync(input.directory, { withFileTypes: true });439 entries = await NodeFS.promises.readdir(input.directory, { withFileTypes: true });
431 } catch (cause) {440 } catch (cause) {
432 return { failures: [{ filePath: input.directory, cause }] };441 return { failures: [{ filePath: input.directory, cause }] };
433 }442 }
...@@ -436,8 +445,8 @@ function enforceRetention(input: {...@@ -436,8 +445,8 @@ function enforceRetention(input: {
436 if (!entry.isFile()) continue;445 if (!entry.isFile()) continue;
437 const filePath = NodePath.join(input.directory, entry.name);446 const filePath = NodePath.join(input.directory, entry.name);
438 try {447 try {
439 if (!isProviderLogFile(filePath, entry.name, input.filePrefix)) continue;448 if (!(await isProviderLogFile(filePath, entry.name, input.filePrefix))) continue;
440 const stat = NodeFS.statSync(filePath);449 const stat = await NodeFS.promises.stat(filePath);
441 files.push({ filePath, mtimeMs: stat.mtimeMs, size: stat.size });450 files.push({ filePath, mtimeMs: stat.mtimeMs, size: stat.size });
442 } catch (cause) {451 } catch (cause) {
443 failures.push({ filePath, cause });452 failures.push({ filePath, cause });
...@@ -445,10 +454,10 @@ function enforceRetention(input: {...@@ -445,10 +454,10 @@ function enforceRetention(input: {
445 }454 }
446455
447 let totalBytes = files.reduce((total, file) => total + file.size, 0);456 let totalBytes = files.reduce((total, file) => total + file.size, 0);
448 const remove = (file: (typeof files)[number]) => {457 const remove = async (file: (typeof files)[number]) => {
449 if (input.activeFilePaths.has(file.filePath)) return false;458 if (input.activeFilePaths.has(file.filePath)) return false;
450 try {459 try {
451 NodeFS.rmSync(file.filePath, { force: true });460 await NodeFS.promises.rm(file.filePath, { force: true });
452 totalBytes -= file.size;461 totalBytes -= file.size;
453 return true;462 return true;
454 } catch (cause) {463 } catch (cause) {
...@@ -457,16 +466,16 @@ function enforceRetention(input: {...@@ -457,16 +466,16 @@ function enforceRetention(input: {
457 }466 }
458 };467 };
459468
460 const retained = files.filter((file) => {469 const retained: typeof files = [];
461 if (input.now - file.mtimeMs <= input.maxAgeMs) return true;470 for (const file of files) {
462 return !remove(file);471 if (input.now - file.mtimeMs <= input.maxAgeMs || !(await remove(file))) retained.push(file);
463 });472 }
464473
465 for (const file of retained.toSorted(474 for (const file of retained.toSorted(
466 (left, right) => left.mtimeMs - right.mtimeMs || left.filePath.localeCompare(right.filePath),475 (left, right) => left.mtimeMs - right.mtimeMs || left.filePath.localeCompare(right.filePath),
467 )) {476 )) {
468 if (totalBytes <= input.maxTotalBytes) break;477 if (totalBytes <= input.maxTotalBytes) break;
469 remove(file);478 await remove(file);
470 }479 }
471480
472 return { failures };481 return { failures };
...@@ -517,7 +526,7 @@ function resolveOptions(...@@ -517,7 +526,7 @@ function resolveOptions(
517 return Effect.succeed(resolved);526 return Effect.succeed(resolved);
518}527}
519528
520function drainPending(input: {529const drainPending = Effect.fnUntraced(function* (input: {
521 readonly directory: string;530 readonly directory: string;
522 readonly options: ResolvedOptions;531 readonly options: ResolvedOptions;
523 readonly state: StoreState;532 readonly state: StoreState;
...@@ -525,7 +534,10 @@ function drainPending(input: {...@@ -525,7 +534,10 @@ function drainPending(input: {
525 readonly now: number;534 readonly now: number;
526 readonly timerFired: boolean;535 readonly timerFired: boolean;
527 readonly close: boolean;536 readonly close: boolean;
528}): readonly [DrainResult, StoreState] {537 readonly makeSink: (
538 filePath: string,
539 ) => Effect.Effect<AsyncFileSink, Effect.Error<ReturnType<typeof makeAsyncRotatingFileSink>>>;
540}): Effect.fn.Return<readonly [DrainResult, StoreState]> {
529 if (input.state.closed) {541 if (input.state.closed) {
530 return [{ attributions: [], failures: [] }, input.state];542 return [{ attributions: [], failures: [] }, input.state];
531 }543 }
...@@ -548,22 +560,20 @@ function drainPending(input: {...@@ -548,22 +560,20 @@ function drainPending(input: {
548 const filePath = providerLogPath(input.directory, input.filePrefix, threadSegment);560 const filePath = providerLogPath(input.directory, input.filePrefix, threadSegment);
549 let sink = sinks.get(threadSegment);561 let sink = sinks.get(threadSegment);
550 if (!sink) {562 if (!sink) {
551 try {563 const created = yield* input.makeSink(filePath).pipe(Effect.result);
552 sink = new RotatingFileSink({564 if (Result.isFailure(created)) {
553 filePath,565 failures.push({ filePath, cause: created.failure });
554 maxBytes: input.options.maxBytes,
555 maxFiles: input.options.maxFiles,
556 throwOnError: true,
557 });
558 sinks.set(threadSegment, sink);
559 } catch (cause) {
560 failures.push({ filePath, cause });
561 continue;566 continue;
562 }567 }
568 sink = created.success;
569 sinks.set(threadSegment, sink);
563 }570 }
564571
565 try {572 const written = yield* writeBatchedMessages(
566 writeBatchedMessages(sink, records, input.options.maxBytes, (writtenRecords) => {573 sink,
574 records,
575 input.options.maxBytes,
576 (writtenRecords) => {
567 for (const record of writtenRecords) {577 for (const record of writtenRecords) {
568 const current = attributionByStream.get(record.stream) ?? {578 const current = attributionByStream.get(record.stream) ?? {
569 count: 0,579 count: 0,
...@@ -574,28 +584,31 @@ function drainPending(input: {...@@ -574,28 +584,31 @@ function drainPending(input: {
574 logicalWriteBytes: current.logicalWriteBytes + record.bytes,584 logicalWriteBytes: current.logicalWriteBytes + record.bytes,
575 });585 });
576 }586 }
577 });587 },
578 } catch (cause) {588 ).pipe(Effect.result);
589 if (Result.isFailure(written)) {
579 sinks.delete(threadSegment);590 sinks.delete(threadSegment);
580 failures.push({ filePath, cause });591 failures.push({ filePath, cause: written.failure });
581 }592 }
582 }593 }
583594
584 const retentionDue =595 const retentionDue =
585 input.now - input.state.lastRetentionAt >= input.options.retentionCheckIntervalMs;596 input.now - input.state.lastRetentionAt >= input.options.retentionCheckIntervalMs;
586 const retention = retentionDue597 const retention = retentionDue
587 ? enforceRetention({598 ? yield* Effect.promise(() =>
588 directory: input.directory,599 enforceRetention({
589 maxTotalBytes: input.options.maxTotalBytes,600 directory: input.directory,
590 maxAgeMs: input.options.maxAgeMs,601 maxTotalBytes: input.options.maxTotalBytes,
591 activeFilePaths: new Set(602 maxAgeMs: input.options.maxAgeMs,
592 Array.from(sinks.keys(), (threadSegment) =>603 activeFilePaths: new Set(
593 providerLogPath(input.directory, input.filePrefix, threadSegment),604 Array.from(sinks.keys(), (threadSegment) =>
605 providerLogPath(input.directory, input.filePrefix, threadSegment),
606 ),
594 ),607 ),
595 ),608 filePrefix: input.filePrefix,
596 filePrefix: input.filePrefix,609 now: input.now,
597 now: input.now,610 }),
598 })611 )
599 : { failures: [] };612 : { failures: [] };
600613
601 return [614 return [
...@@ -615,7 +628,7 @@ function drainPending(input: {...@@ -615,7 +628,7 @@ function drainPending(input: {
615 lastRetentionAt: retentionDue ? input.now : input.state.lastRetentionAt,628 lastRetentionAt: retentionDue ? input.now : input.state.lastRetentionAt,
616 },629 },
617 ];630 ];
618}631});
619632
620const serializeEvent = Effect.fnUntraced(function* (event: unknown) {633const serializeEvent = Effect.fnUntraced(function* (event: unknown) {
621 return yield* encodeUnknownJsonString(event).pipe(634 return yield* encodeUnknownJsonString(event).pipe(
...@@ -630,18 +643,32 @@ const serializeEvent = Effect.fnUntraced(function* (event: unknown) {...@@ -630,18 +643,32 @@ const serializeEvent = Effect.fnUntraced(function* (event: unknown) {
630export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (643export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
631 filePath: string,644 filePath: string,
632 options: EventNdjsonLogStoreOptions = {},645 options: EventNdjsonLogStoreOptions = {},
633): Effect.fn.Return<EventNdjsonLogStore, EventNdjsonLogStoreError> {646): Effect.fn.Return<
647 EventNdjsonLogStore,
648 EventNdjsonLogStoreError,
649 FileSystem.FileSystem | Path.Path
650> {
651 const fs = yield* FileSystem.FileSystem;
652 const path = yield* Path.Path;
634 const resolved = yield* resolveOptions(filePath, options);653 const resolved = yield* resolveOptions(filePath, options);
635 const directory = NodePath.dirname(filePath);654 const directory = NodePath.dirname(filePath);
636 const filePrefix = providerLogPrefix(filePath);655 const filePrefix = providerLogPrefix(filePath);
637656
638 yield* Effect.try({657 yield* fs
639 try: () => NodeFS.mkdirSync(directory, { recursive: true }),658 .makeDirectory(directory, { recursive: true })
640 catch: (cause) => new EventNdjsonLogDirectoryError({ directory, cause }),659 .pipe(Effect.mapError((cause) => new EventNdjsonLogDirectoryError({ directory, cause })));
641 });660 const makeSink = (filePath: string) =>
661 makeAsyncRotatingFileSink({
662 filePath,
663 maxBytes: resolved.maxBytes,
664 maxFiles: resolved.maxFiles,
665 }).pipe(
666 Effect.provideService(FileSystem.FileSystem, fs),
667 Effect.provideService(Path.Path, path),
668 );
642669
643 const initializedAt = yield* Clock.currentTimeMillis;670 const initializedAt = yield* Clock.currentTimeMillis;
644 const initialRetention = yield* Effect.sync(() =>671 const initialRetention = yield* Effect.promise(() =>
645 enforceRetention({672 enforceRetention({
646 directory,673 directory,
647 maxTotalBytes: resolved.maxTotalBytes,674 maxTotalBytes: resolved.maxTotalBytes,
...@@ -671,17 +698,16 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (...@@ -671,17 +698,16 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
671 const flush = Effect.fnUntraced(function* (timerFired: boolean, close: boolean) {698 const flush = Effect.fnUntraced(function* (timerFired: boolean, close: boolean) {
672 const startedAt = yield* Clock.currentTimeMillis;699 const startedAt = yield* Clock.currentTimeMillis;
673 const result = yield* SynchronizedRef.modifyEffect(stateRef, (state) =>700 const result = yield* SynchronizedRef.modifyEffect(stateRef, (state) =>
674 Effect.sync(() =>701 drainPending({
675 drainPending({702 directory,
676 directory,703 options: resolved,
677 options: resolved,704 state,
678 state,705 filePrefix,
679 filePrefix,706 now: startedAt,
680 now: startedAt,707 timerFired,
681 timerFired,708 close,
682 close,709 makeSink,
683 }),710 }),
684 ),
685 );711 );
686712
687 for (const failure of result.failures) {713 for (const failure of result.failures) {
...@@ -714,7 +740,7 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (...@@ -714,7 +740,7 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
714 { discard: true },740 { discard: true },
715 );741 );
716 }742 }
717 });743 }, Effect.uninterruptible);
718744
719 const scheduleFlush = Effect.fnUntraced(function* () {745 const scheduleFlush = Effect.fnUntraced(function* () {
720 yield* Effect.forkIn(746 yield* Effect.forkIn(
...@@ -791,7 +817,11 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (...@@ -791,7 +817,11 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
791export const makeEventNdjsonLogger = Effect.fnUntraced(function* (817export const makeEventNdjsonLogger = Effect.fnUntraced(function* (
792 filePath: string,818 filePath: string,
793 options: EventNdjsonLoggerOptions,819 options: EventNdjsonLoggerOptions,
794): Effect.fn.Return<ProviderEventLoggers.EventNdjsonLogger | undefined> {820): Effect.fn.Return<
821 ProviderEventLoggers.EventNdjsonLogger | undefined,
822 never,
823 FileSystem.FileSystem | Path.Path
824> {
795 const store = yield* makeEventNdjsonLogStore(filePath, options).pipe(825 const store = yield* makeEventNdjsonLogStore(filePath, options).pipe(
796 Effect.catch((error) =>826 Effect.catch((error) =>
797 logWarning(error.message, { error }).pipe(827 logWarning(error.message, { error }).pipe(
apps/server/src/serverRuntimeStartup.ts+15-9
...@@ -346,8 +346,10 @@ const maybeOpenBrowser = (target: string) =>...@@ -346,8 +346,10 @@ const maybeOpenBrowser = (target: string) =>
346346
347const runStartupPhase = <A, E, R>(phase: string, effect: Effect.Effect<A, E, R>) =>347const runStartupPhase = <A, E, R>(phase: string, effect: Effect.Effect<A, E, R>) =>
348 effect.pipe(348 effect.pipe(
349 Effect.annotateSpans({ "startup.phase": phase }),349 // Attributes on the span itself, not annotateSpans: an annotation follows
350 Effect.withSpan(`server.startup.${phase}`),350 // every fiber forked inside the phase, such as the effect worker, for the
351 // life of the server.
352 Effect.withSpan(`server.startup.${phase}`, { attributes: { "startup.phase": phase } }),
351 );353 );
352354
353interface StartupOptions {355interface StartupOptions {
...@@ -581,8 +583,9 @@ const make = (options?: StartupOptions) =>...@@ -581,8 +583,9 @@ const make = (options?: StartupOptions) =>
581 Effect.gen(function* () {583 Effect.gen(function* () {
582 yield* Effect.logDebug("startup phase: recording startup heartbeat");584 yield* Effect.logDebug("startup phase: recording startup heartbeat");
583 yield* recordStartupHeartbeat.pipe(585 yield* recordStartupHeartbeat.pipe(
584 Effect.annotateSpans({ "startup.phase": "heartbeat.record" }),586 Effect.withSpan("server.startup.heartbeat.record", {
585 Effect.withSpan("server.startup.heartbeat.record"),587 attributes: { "startup.phase": "heartbeat.record" },
588 }),
586 Effect.ignoreCause({ log: true }),589 Effect.ignoreCause({ log: true }),
587 );590 );
588 if (serverConfig.startupPresentation === "headless") {591 if (serverConfig.startupPresentation === "headless") {
...@@ -653,12 +656,15 @@ const make = (options?: StartupOptions) =>...@@ -653,12 +656,15 @@ const make = (options?: StartupOptions) =>
653 yield* Effect.logDebug("startup phase: complete");656 yield* Effect.logDebug("startup phase: complete");
654 yield* flushCompileCache;657 yield* flushCompileCache;
655 }).pipe(658 }).pipe(
656 Effect.annotateSpans({659 Effect.withSpan("server.startup", {
657 "server.mode": serverConfig.mode,660 kind: "server",
658 "server.port": serverConfig.port,661 root: true,
659 "server.host": serverConfig.host ?? "default",662 attributes: {
663 "server.mode": serverConfig.mode,
664 "server.port": serverConfig.port,
665 "server.host": serverConfig.host ?? "default",
666 },
660 }),667 }),
661 Effect.withSpan("server.startup", { kind: "server", root: true }),
662 );668 );
663669
664 yield* Effect.forkScoped(670 yield* Effect.forkScoped(
docs/internals/performance-regressions.md+6
...@@ -90,6 +90,12 @@ counts keep the full status path. Cached fetch failures share the existing backo...@@ -90,6 +90,12 @@ counts keep the full status path. Cached fetch failures share the existing backo
90the actual failed attempt. Unchanged desktop environment bootstrap reads retain their array90the actual failed attempt. Unchanged desktop environment bootstrap reads retain their array
91identity so polling does not invalidate subscribers.91identity so polling does not invalidate subscribers.
9292
93Local trace records and provider event logs use asynchronous, serialized file writes and rotation.
94Batching synchronous appends still stalls the server event loop on slow storage, especially on
95Windows. While a trace write is slow, producers keep up to one trace file of records, so the spans
96that time a stall survive it; records drop only while the disk stays stuck. Provider logging
97retains backpressure. Shutdown drains accepted writes before closing their scope.
98
93Relay awareness queues one pending publication per thread and retains a follow-up when the thread99Relay awareness queues one pending publication per thread and retains a follow-up when the thread
94changes during a send. Transcript deltas and tool output do not enqueue awareness work. Run,100changes during a send. Transcript deltas and tool output do not enqueue awareness work. Run,
95request, and relevant thread metadata events do. Configuration is checked before shell reads, and101request, and relevant thread metadata events do. Configuration is checked before shell reads, and
docs/operations/observability.md+2-1
...@@ -646,7 +646,8 @@ Current high-value span and metric boundaries include:...@@ -646,7 +646,8 @@ Current high-value span and metric boundaries include:
646- provider session and turn operations646- provider session and turn operations
647- git command execution and git hook events647- git command execution and git hook events
648- terminal session lifecycle648- terminal session lifecycle
649- sqlite query execution649- sqlite query execution, except the statements inside `orchestrationV2.EventSink.write`, which
650 its `sql.transaction` span times as one
650- event loop stalls (`server.eventLoop.stall`)651- event loop stalls (`server.eventLoop.stall`)
651652
652### Current Constraints653### Current Constraints
packages/shared/src/logging.test.ts+61
...@@ -3,12 +3,17 @@ import * as NodeFS from "node:fs";...@@ -3,12 +3,17 @@ import * as NodeFS from "node:fs";
3import * as NodeOS from "node:os";3import * as NodeOS from "node:os";
4import * as NodePath from "node:path";4import * as NodePath from "node:path";
5import { afterEach, describe, expect, it } from "vite-plus/test";5import { afterEach, describe, expect, it } from "vite-plus/test";
6import * as NodeServices from "@effect/platform-node/NodeServices";
7import { it as effectIt } from "@effect/vitest";
8import * as Effect from "effect/Effect";
9import * as FileSystem from "effect/FileSystem";
6import * as HostProcess from "./HostProcess.ts";10import * as HostProcess from "./HostProcess.ts";
711
8import {12import {
9 RotatingFileSink,13 RotatingFileSink,
10 RotatingFileSinkConfigurationError,14 RotatingFileSinkConfigurationError,
11 RotatingFileSinkError,15 RotatingFileSinkError,
16 makeAsyncRotatingFileSink,
12} from "./logging.ts";17} from "./logging.ts";
1318
14const windowsHost = HostProcess.Platform.defaultValue() === "win32";19const windowsHost = HostProcess.Platform.defaultValue() === "win32";
...@@ -35,6 +40,62 @@ afterEach(() => {...@@ -35,6 +40,62 @@ afterEach(() => {
35 }40 }
36});41});
3742
43effectIt.layer(NodeServices.layer)("asynchronous rotating file sink", (it) => {
44 it.effect("serializes concurrent appends across rotation and resumes an existing file", () =>
45 Effect.gen(function* () {
46 const directory = makeTempDirectory();
47 const filePath = NodePath.join(directory, "async.log");
48 NodeFS.writeFileSync(`${filePath}.50`, "obsolete");
49 const lines = Array.from(
50 { length: 100 },
51 (_, index) => `${index.toString().padStart(3, "0")}\n`,
52 );
53 const sink = yield* makeAsyncRotatingFileSink({ filePath, maxBytes: 32, maxFiles: 20 });
54 yield* Effect.forEach(lines, (line) => sink.write(line), {
55 concurrency: "unbounded",
56 discard: true,
57 });
58 const resumed = yield* makeAsyncRotatingFileSink({ filePath, maxBytes: 32, maxFiles: 20 });
59 yield* resumed.write("100\n");
60 expect(NodeFS.existsSync(`${filePath}.50`)).toBe(false);
61 const backups = NodeFS.readdirSync(directory)
62 .filter((name) => name.startsWith("async.log."))
63 .sort((a, b) => Number(b.slice(10)) - Number(a.slice(10)));
64 const contents = [...backups, "async.log"].map((name) =>
65 NodeFS.readFileSync(NodePath.join(directory, name), "utf8"),
66 );
67 expect(contents.every((chunk) => Buffer.byteLength(chunk) <= 32)).toBe(true);
68 expect(contents.join("")).toBe(lines.join("") + "100\n");
69 }),
70 );
71
72 it.effect("reports a failed append and recovers its size before the next rotation", () =>
73 Effect.gen(function* () {
74 const directory = makeTempDirectory();
75 const filePath = NodePath.join(directory, "recovery.log");
76 NodeFS.writeFileSync(filePath, "old\n");
77 const fs = yield* FileSystem.FileSystem;
78 let attempts = 0;
79 const sink = yield* makeAsyncRotatingFileSink({ filePath, maxBytes: 8, maxFiles: 2 }).pipe(
80 Effect.provideService(FileSystem.FileSystem, {
81 ...fs,
82 writeFile: (path, bytes, options) =>
83 ++attempts === 1
84 ? fs.writeFile(directory, bytes, options)
85 : fs.writeFile(path, bytes, options),
86 }),
87 );
88 const failed = yield* sink.write("new\n").pipe(Effect.result);
89 expect(failed._tag).toBe("Failure");
90 if (failed._tag === "Failure") expect(failed.failure.operation).toBe("write");
91 yield* sink.write("new\n");
92 yield* sink.write("end\n");
93 expect(NodeFS.readFileSync(`${filePath}.1`, "utf8")).toBe("old\nnew\n");
94 expect(NodeFS.readFileSync(filePath, "utf8")).toBe("end\n");
95 }),
96 );
97});
98
38describe("RotatingFileSink", () => {99describe("RotatingFileSink", () => {
39 it.each([100 it.each([
40 { option: "maxBytes" as const, maxBytes: 0, maxFiles: 1 },101 { option: "maxBytes" as const, maxBytes: 0, maxFiles: 1 },
packages/shared/src/logging.ts+83
...@@ -2,6 +2,10 @@...@@ -2,6 +2,10 @@
2import * as NodeFS from "node:fs";2import * as NodeFS from "node:fs";
3import * as NodePath from "node:path";3import * as NodePath from "node:path";
4import * as Schema from "effect/Schema";4import * as Schema from "effect/Schema";
5import * as Effect from "effect/Effect";
6import * as FileSystem from "effect/FileSystem";
7import * as Path from "effect/Path";
8import * as Semaphore from "effect/Semaphore";
59
6export interface RotatingFileSinkOptions {10export interface RotatingFileSinkOptions {
7 readonly filePath: string;11 readonly filePath: string;
...@@ -180,3 +184,82 @@ export class RotatingFileSink {...@@ -180,3 +184,82 @@ export class RotatingFileSink {
180 return `${this.filePath}.${index}`;184 return `${this.filePath}.${index}`;
181 }185 }
182}186}
187
188/** Serialized asynchronous writes and rotation without blocking the event loop. */
189export const makeAsyncRotatingFileSink = Effect.fnUntraced(function* (
190 options: Omit<RotatingFileSinkOptions, "throwOnError">,
191) {
192 const fs = yield* FileSystem.FileSystem;
193 const path = yield* Path.Path;
194 if (options.maxBytes < 1)
195 return yield* new RotatingFileSinkConfigurationError({
196 option: "maxBytes",
197 received: options.maxBytes,
198 minimum: 1,
199 });
200 if (options.maxFiles < 1)
201 return yield* new RotatingFileSinkConfigurationError({
202 option: "maxFiles",
203 received: options.maxFiles,
204 minimum: 1,
205 });
206 const failure = (operation: RotatingFileSinkError["operation"]) => (cause: unknown) =>
207 new RotatingFileSinkError({ operation, filePath: options.filePath, cause });
208 const directory = path.dirname(options.filePath);
209 const basename = path.basename(options.filePath);
210 yield* fs
211 .makeDirectory(directory, { recursive: true })
212 .pipe(Effect.mapError(failure("initialize")));
213 yield* Effect.gen(function* () {
214 for (const entry of yield* fs.readDirectory(directory)) {
215 if (!entry.startsWith(`${basename}.`)) continue;
216 const suffix = Number(entry.slice(basename.length + 1));
217 if (!Number.isInteger(suffix) || suffix <= options.maxFiles) continue;
218 yield* fs.remove(path.join(directory, entry), { force: true });
219 }
220 }).pipe(Effect.mapError(failure("prune")));
221 const readSize = fs.stat(options.filePath).pipe(
222 Effect.map((stat) => Number(stat.size)),
223 Effect.catch((cause) =>
224 cause.reason._tag === "NotFound" ? Effect.succeed(0) : Effect.fail(failure("read")(cause)),
225 ),
226 );
227 let currentSize = yield* readSize;
228 const semaphore = yield* Semaphore.make(1);
229 const rotate = Effect.gen(function* () {
230 yield* fs.remove(`${options.filePath}.${options.maxFiles}`, { force: true });
231 for (let index = options.maxFiles - 1; index >= 1; index--) {
232 const source = `${options.filePath}.${index}`;
233 if (yield* fs.exists(source)) yield* fs.rename(source, `${options.filePath}.${index + 1}`);
234 }
235 if (yield* fs.exists(options.filePath))
236 yield* fs.rename(options.filePath, `${options.filePath}.1`);
237 currentSize = 0;
238 }).pipe(Effect.mapError(failure("rotate")));
239 const write = (chunk: string | Buffer) =>
240 semaphore
241 .withPermits(1)(
242 Effect.gen(function* () {
243 const buffer = typeof chunk === "string" ? Buffer.from(chunk) : chunk;
244 if (buffer.length === 0) return;
245 if (currentSize > 0 && currentSize + buffer.length > options.maxBytes) yield* rotate;
246 yield* fs
247 .writeFile(options.filePath, buffer, { flag: "a" })
248 .pipe(Effect.mapError(failure("write")));
249 currentSize += buffer.length;
250 }).pipe(
251 Effect.tapError(() =>
252 readSize.pipe(
253 Effect.tap((size) =>
254 Effect.sync(() => {
255 currentSize = size;
256 }),
257 ),
258 Effect.ignore,
259 ),
260 ),
261 ),
262 )
263 .pipe(Effect.uninterruptible);
264 return { write };
265});
packages/shared/src/observability.test.ts+70-6
...@@ -3,6 +3,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices";...@@ -3,6 +3,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices";
3import * as Arr from "effect/Array";3import * as Arr from "effect/Array";
4import * as Cause from "effect/Cause";4import * as Cause from "effect/Cause";
5import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
6import * as Deferred from "effect/Deferred";
7import * as Fiber from "effect/Fiber";
6import * as Exit from "effect/Exit";8import * as Exit from "effect/Exit";
7import * as FileSystem from "effect/FileSystem";9import * as FileSystem from "effect/FileSystem";
8import * as Layer from "effect/Layer";10import * as Layer from "effect/Layer";
...@@ -16,7 +18,6 @@ import * as Tracer from "effect/Tracer";...@@ -16,7 +18,6 @@ import * as Tracer from "effect/Tracer";
16import * as TestClock from "effect/testing/TestClock";18import * as TestClock from "effect/testing/TestClock";
17import { vi } from "vite-plus/test";19import { vi } from "vite-plus/test";
1820
19import { RotatingFileSink } from "./logging.ts";
20import {21import {
21 causeErrorTag,22 causeErrorTag,
22 compactTraceAttributes,23 compactTraceAttributes,
...@@ -242,7 +243,7 @@ describe("observability", () => {...@@ -242,7 +243,7 @@ describe("observability", () => {
242 sampled ? [{ _tag: "Success" }] : [],243 sampled ? [{ _tag: "Success" }] : [],
243 );244 );
244 }245 }
245 }),246 }).pipe(Effect.provide(NodeServices.layer)),
246 );247 );
247248
248 it.effect("preserves failure and interruption causes in spans and exported traces", () =>249 it.effect("preserves failure and interruption causes in spans and exported traces", () =>
...@@ -290,7 +291,7 @@ describe("observability", () => {...@@ -290,7 +291,7 @@ describe("observability", () => {
290 }291 }
291 assert.equal(records[0]?.exit._tag, Cause.hasInterrupts(cause) ? "Interrupted" : "Failure");292 assert.equal(records[0]?.exit._tag, Cause.hasInterrupts(cause) ? "Interrupted" : "Failure");
292 }293 }
293 }),294 }).pipe(Effect.provide(NodeServices.layer)),
294 );295 );
295296
296 it("normalizes circular arrays, maps, and sets without recursing forever", () => {297 it("normalizes circular arrays, maps, and sets without recursing forever", () => {
...@@ -331,6 +332,64 @@ describe("observability", () => {...@@ -331,6 +332,64 @@ describe("observability", () => {
331 });332 });
332333
333 nodeServicesIt("node services", (it) => {334 nodeServicesIt("node services", (it) => {
335 it.effect(
336 "keeps trace producers responsive during a slow write, bounds the backlog, and drains on close",
337 () =>
338 Effect.scoped(
339 Effect.gen(function* () {
340 const fs = yield* FileSystem.FileSystem;
341 const path = yield* Path.Path;
342 const directory = yield* fs.makeTempDirectoryScoped({
343 prefix: "t3-trace-backpressure-",
344 });
345 const tracePath = path.join(directory, "trace.ndjson");
346 const started = yield* Deferred.make<void>();
347 const release = yield* Deferred.make<void>();
348 const closed = yield* Ref.make(false);
349 const sink = yield* makeTraceSink({
350 filePath: tracePath,
351 maxBytes: 5_000_000,
352 maxFiles: 2,
353 batchWindowMs: 10_000,
354 }).pipe(
355 Effect.provideService(FileSystem.FileSystem, {
356 ...fs,
357 writeFile: (file, data, options) =>
358 Effect.gen(function* () {
359 yield* Deferred.succeed(started, undefined);
360 yield* Deferred.await(release);
361 yield* fs.writeFile(file, data, options);
362 }),
363 }),
364 );
365 sink.push(makeRecord("first"));
366 const flushing = yield* Effect.forkChild(sink.flush);
367 yield* Deferred.await(started);
368 // These pushes return while the write is held. The backlog keeps one
369 // trace file of records: every small one, then large ones until the
370 // next would not fit.
371 for (let index = 0; index < 2_000; index++)
372 sink.push(makeRecord("small", String(index)));
373 for (let index = 0; index < 3; index++)
374 sink.push({ ...makeRecord("large"), attributes: { payload: "x".repeat(2_000_000) } });
375 const closing = yield* Effect.forkChild(
376 sink.close().pipe(Effect.tap(() => Ref.set(closed, true))),
377 );
378 yield* Effect.yieldNow;
379 assert.isFalse(yield* Ref.get(closed));
380 yield* Deferred.succeed(release, undefined);
381 yield* Fiber.join(flushing);
382 yield* Fiber.join(closing);
383 const records = yield* readTraceRecords(tracePath);
384 assert.deepStrictEqual(
385 records.map((record) => record.name),
386 ["first", ...Array.from({ length: 2_000 }, () => "small"), "large", "large"],
387 );
388 assert.isTrue(yield* Ref.get(closed));
389 }),
390 ),
391 );
392
334 it.effect("flushes buffered trace records on close", () =>393 it.effect("flushes buffered trace records on close", () =>
335 Effect.scoped(394 Effect.scoped(
336 Effect.gen(function* () {395 Effect.gen(function* () {
...@@ -551,7 +610,7 @@ describe("observability", () => {...@@ -551,7 +610,7 @@ describe("observability", () => {
551 const tracePath = path.join(tempDir, "shared.trace.ndjson");610 const tracePath = path.join(tempDir, "shared.trace.ndjson");
552 // A directory at the trace path fails every append, like a full disk.611 // A directory at the trace path fails every append, like a full disk.
553 yield* fileSystem.makeDirectory(tracePath);612 yield* fileSystem.makeDirectory(tracePath);
554 const write = vi.spyOn(RotatingFileSink.prototype, "write");613 const write = vi.spyOn(fileSystem, "writeFile");
555 yield* Effect.addFinalizer(() => Effect.sync(() => write.mockRestore()));614 yield* Effect.addFinalizer(() => Effect.sync(() => write.mockRestore()));
556615
557 const sink = yield* makeTraceSink({616 const sink = yield* makeTraceSink({
...@@ -564,16 +623,18 @@ describe("observability", () => {...@@ -564,16 +623,18 @@ describe("observability", () => {
564 for (let index = 0; index < 1_024; index += 1) {623 for (let index = 0; index < 1_024; index += 1) {
565 sink.push(makeRecord("lost", String(index)));624 sink.push(makeRecord("lost", String(index)));
566 }625 }
626 yield* sink.flush;
567 // Timed flushes run in the fiber forked inside the makeTraceSink span.627 // Timed flushes run in the fiber forked inside the makeTraceSink span.
568 for (let index = 0; index < 5; index += 1) {628 for (let index = 0; index < 5; index += 1) {
569 sink.push(makeRecord("lost"));629 sink.push(makeRecord("lost"));
570 yield* TestClock.adjust("1 second");630 yield* TestClock.adjust("1 second");
631 yield* sink.flush;
571 }632 }
572633
573 // One write per batch, never a growing backlog.634 // One write per batch, never a growing backlog.
574 assert.deepStrictEqual(635 assert.deepStrictEqual(
575 write.mock.calls.map(([chunk]) => String(chunk).split("\n").length - 1),636 write.mock.calls.map(([, chunk]) => Buffer.from(chunk).toString().split("\n").length - 1),
576 [256, 256, 256, 256, 1, 1, 1, 1, 1],637 [1024, 1, 1, 1, 1, 1],
577 );638 );
578 expect(logs).toEqual([639 expect(logs).toEqual([
579 { logLevel: "Warn", message: [expect.any(String), { filePath: tracePath }] },640 { logLevel: "Warn", message: [expect.any(String), { filePath: tracePath }] },
...@@ -583,6 +644,7 @@ describe("observability", () => {...@@ -583,6 +644,7 @@ describe("observability", () => {
583 yield* fileSystem.remove(tracePath, { recursive: true });644 yield* fileSystem.remove(tracePath, { recursive: true });
584 sink.push(makeRecord("recovered"));645 sink.push(makeRecord("recovered"));
585 yield* TestClock.adjust("1 second");646 yield* TestClock.adjust("1 second");
647 yield* sink.flush;
586648
587 const records = yield* readTraceRecords(tracePath);649 const records = yield* readTraceRecords(tracePath);
588 assert.deepStrictEqual(650 assert.deepStrictEqual(
...@@ -597,6 +659,7 @@ describe("observability", () => {...@@ -597,6 +659,7 @@ describe("observability", () => {
597 // Healthy flushes after the recovery log nothing.659 // Healthy flushes after the recovery log nothing.
598 sink.push(makeRecord("healthy"));660 sink.push(makeRecord("healthy"));
599 yield* TestClock.adjust("1 second");661 yield* TestClock.adjust("1 second");
662 yield* sink.flush;
600 expect(logs).toHaveLength(2);663 expect(logs).toHaveLength(2);
601664
602 // A new failure episode warns again.665 // A new failure episode warns again.
...@@ -604,6 +667,7 @@ describe("observability", () => {...@@ -604,6 +667,7 @@ describe("observability", () => {
604 yield* fileSystem.makeDirectory(tracePath);667 yield* fileSystem.makeDirectory(tracePath);
605 sink.push(makeRecord("lost-again"));668 sink.push(makeRecord("lost-again"));
606 yield* TestClock.adjust("1 second");669 yield* TestClock.adjust("1 second");
670 yield* sink.flush;
607 assert.deepStrictEqual(671 assert.deepStrictEqual(
608 logs.map((log) => log.logLevel),672 logs.map((log) => log.logLevel),
609 ["Warn", "Info", "Warn"],673 ["Warn", "Info", "Warn"],
packages/shared/src/observability.ts+84-54
...@@ -4,13 +4,16 @@ import * as Effect from "effect/Effect";...@@ -4,13 +4,16 @@ import * as Effect from "effect/Effect";
4import type * as Exit from "effect/Exit";4import type * as Exit from "effect/Exit";
5import * as ExitRuntime from "effect/Exit";5import * as ExitRuntime from "effect/Exit";
6import * as Option from "effect/Option";6import * as Option from "effect/Option";
7import * as Queue from "effect/Queue";
8import * as Result from "effect/Result";
9import * as Semaphore from "effect/Semaphore";
7import * as Schema from "effect/Schema";10import * as Schema from "effect/Schema";
8import * as SchemaIssue from "effect/SchemaIssue";11import * as SchemaIssue from "effect/SchemaIssue";
9import * as SchemaTransformation from "effect/SchemaTransformation";12import * as SchemaTransformation from "effect/SchemaTransformation";
10import * as Tracer from "effect/Tracer";13import * as Tracer from "effect/Tracer";
11import { OtlpResource, OtlpTracer, OtlpSerialization } from "effect/observability";14import { OtlpResource, OtlpTracer, OtlpSerialization } from "effect/observability";
1215
13import { RotatingFileSink } from "./logging.ts";16import { makeAsyncRotatingFileSink } from "./logging.ts";
1417
15export const OtlpProtocol = Schema.Literals(["http/json", "http/protobuf"]);18export const OtlpProtocol = Schema.Literals(["http/json", "http/protobuf"]);
16export type OtlpProtocol = typeof OtlpProtocol.Type;19export type OtlpProtocol = typeof OtlpProtocol.Type;
...@@ -37,7 +40,6 @@ export const DEFAULT_SIGNAL_EXPORT: SignalExport = {...@@ -37,7 +40,6 @@ export const DEFAULT_SIGNAL_EXPORT: SignalExport = {
37};40};
3841
39const FLUSH_BUFFER_THRESHOLD = 256;42const FLUSH_BUFFER_THRESHOLD = 256;
40const textEncoder = new TextEncoder();
4143
42export type TraceAttributes = Readonly<Record<string, unknown>>;44export type TraceAttributes = Readonly<Record<string, unknown>>;
4345
...@@ -364,14 +366,20 @@ function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord {...@@ -364,14 +366,20 @@ function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord {
364}366}
365367
366export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) {368export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) {
367 const sink = new RotatingFileSink({369 const sink = yield* makeAsyncRotatingFileSink({
368 filePath: options.filePath,370 filePath: options.filePath,
369 maxBytes: options.maxBytes,371 maxBytes: options.maxBytes,
370 maxFiles: options.maxFiles,372 maxFiles: options.maxFiles,
371 throwOnError: true,373 }).pipe(Effect.orDie);
372 });374
373375 let buffer: Array<{ readonly text: string; readonly bytes: number }> = [];
374 let buffer: Array<string> = [];376 let bufferBytes = 0;
377 // While a write is slow, push keeps up to one trace file of records (at least
378 // 4 MiB), so the spans that time a stall still reach the file. Only a disk
379 // that stays stuck drops records, instead of growing memory or blocking push.
380 const maxBufferBytes = Math.max(options.maxBytes, 4 * 1024 * 1024);
381 const flushRequests = yield* Queue.sliding<void>(1);
382 const flushSemaphore = yield* Semaphore.make(1);
375 // A failure episode starts at the first dropped record and ends when a flush383 // A failure episode starts at the first dropped record and ends when a flush
376 // sees that the latest write succeeded. Flush checks once per window, so a384 // sees that the latest write succeeded. Flush checks once per window, so a
377 // disk that flaps inside one window stays one episode.385 // disk that flaps inside one window stays one episode.
...@@ -384,36 +392,35 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac...@@ -384,36 +392,35 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac
384 durationMs: 0,392 durationMs: 0,
385 };393 };
386394
387 const flushUnsafe = () => {395 const flushBuffered = Effect.gen(function* () {
388 if (buffer.length === 0) {396 if (buffer.length === 0) {
389 return;397 return;
390 }398 }
391399
392 const records = buffer;400 const records = buffer;
393 buffer = [];401 buffer = [];
402 bufferBytes = 0;
394 let persistedCount = 0;403 let persistedCount = 0;
395404
396 while (persistedCount < records.length) {405 while (persistedCount < records.length) {
397 const firstRecordBytes = textEncoder.encode(records[persistedCount]).byteLength;406 // push drops any record larger than a file, so the first one always fits.
398 if (firstRecordBytes > options.maxBytes) {407 const firstRecordBytes = records[persistedCount]!.bytes;
399 persistedCount += 1;
400 continue;
401 }
402
403 let nextIndex = persistedCount + 1;408 let nextIndex = persistedCount + 1;
404 let chunkBytes = firstRecordBytes;409 let chunkBytes = firstRecordBytes;
405 while (nextIndex < records.length) {410 while (nextIndex < records.length) {
406 const nextRecordBytes = textEncoder.encode(records[nextIndex]).byteLength;411 const nextRecordBytes = records[nextIndex]!.bytes;
407 if (chunkBytes + nextRecordBytes > options.maxBytes) break;412 if (chunkBytes + nextRecordBytes > options.maxBytes) break;
408 chunkBytes += nextRecordBytes;413 chunkBytes += nextRecordBytes;
409 nextIndex += 1;414 nextIndex += 1;
410 }415 }
411416
412 const chunk = records.slice(persistedCount, nextIndex).join("");417 const chunk = records
418 .slice(persistedCount, nextIndex)
419 .map((record) => record.text)
420 .join("");
413 const startedAt = performance.now();421 const startedAt = performance.now();
414 try {422 const written = yield* sink.write(chunk).pipe(Effect.result);
415 sink.write(chunk);423 if (Result.isFailure(written)) {
416 } catch {
417 // A failing disk (ENOSPC, EACCES, EIO) drops the rest of the batch.424 // A failing disk (ENOSPC, EACCES, EIO) drops the rest of the batch.
418 // Retrying it would grow the backlog, and every later push would425 // Retrying it would grow the backlog, and every later push would
419 // retry all of it.426 // retry all of it.
...@@ -429,54 +436,77 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac...@@ -429,54 +436,77 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac
429 };436 };
430 persistedCount = nextIndex;437 persistedCount = nextIndex;
431 }438 }
432 };439 });
433440
434 // Logs once when writes start failing and once when they recover, so a disk441 // Logs once when writes start failing and once when they recover, so a disk
435 // that stays broken costs one line, not one per flush.442 // that stays broken costs one line, not one per flush.
436 const flush = Effect.gen(function* () {443 const flush = flushSemaphore
437 flushUnsafe();444 .withPermits(1)(
438 const stats = pendingFlushStats;445 Effect.gen(function* () {
439 pendingFlushStats = {446 const droppedAtStart = droppedCount;
440 logicalWriteBytes: 0,447 yield* flushBuffered;
441 count: 0,448 const stats = pendingFlushStats;
442 durationMs: 0,449 pendingFlushStats = {
443 };450 logicalWriteBytes: 0,
444 if (stats.count > 0 && options.onFlush) {451 count: 0,
445 yield* options.onFlush(stats).pipe(Effect.ignore);452 durationMs: 0,
446 }453 };
447 if (droppedCount > 0 && !failureReported) {454 if (stats.count > 0 && options.onFlush) {
448 failureReported = true;455 yield* options.onFlush(stats).pipe(Effect.ignore);
449 yield* Effect.logWarning("Trace writes are failing, dropping records until they recover", {456 }
450 filePath: options.filePath,457 if (droppedCount > 0 && !failureReported) {
451 });458 failureReported = true;
452 }459 yield* Effect.logWarning(
453 if (failureReported && !writeFailing) {460 "Trace writer could not keep up, dropping records until it recovers",
454 yield* Effect.logInfo("Trace writes recovered", {461 {
455 filePath: options.filePath,462 filePath: options.filePath,
456 droppedCount,463 },
457 });464 );
458 failureReported = false;465 } else if (failureReported && !writeFailing && droppedCount === droppedAtStart) {
459 droppedCount = 0;466 yield* Effect.logInfo("Trace writes recovered", {
460 }467 filePath: options.filePath,
461 }).pipe(468 droppedCount,
462 Effect.withTracerEnabled(false),469 });
463 // The timed fiber inherits the makeTraceSink span. That span has ended but470 failureReported = false;
464 // lives as long as the sink, so the tracer logger must not add logs to it.471 droppedCount = 0;
465 Effect.updateContext(Context.omit(Tracer.ParentSpan)<never>),472 }
466 );473 }),
474 )
475 .pipe(
476 // A cancelled write still completes in the OS. Finish it before another
477 // flush rotates the file or the scope's finalizer drains the remaining records.
478 Effect.uninterruptible,
479 Effect.withTracerEnabled(false),
480 // The timed fiber inherits the makeTraceSink span. That span has ended but
481 // lives as long as the sink, so the tracer logger must not add logs to it.
482 Effect.updateContext(Context.omit(Tracer.ParentSpan)<never>),
483 );
467484
468 yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore));485 yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore));
469 yield* Effect.forkScoped(486 yield* Effect.forkScoped(
470 Effect.sleep(`${options.batchWindowMs} millis`).pipe(Effect.andThen(flush), Effect.forever),487 // Flush when push fills a batch or the batch window passes.
488 Queue.take(flushRequests).pipe(
489 Effect.timeoutOption(`${options.batchWindowMs} millis`),
490 Effect.andThen(flush),
491 Effect.forever,
492 ),
471 );493 );
472494
473 return {495 return {
474 filePath: options.filePath,496 filePath: options.filePath,
475 push(record) {497 push(record) {
476 try {498 try {
477 buffer.push(`${JSON.stringify(record)}\n`);499 const text = `${JSON.stringify(record)}\n`;
500 const bytes = Buffer.byteLength(text);
501 if (bytes > options.maxBytes) return;
502 if (bufferBytes + bytes > maxBufferBytes) {
503 droppedCount++;
504 return;
505 }
506 buffer.push({ text, bytes });
507 bufferBytes += bytes;
478 if (buffer.length >= FLUSH_BUFFER_THRESHOLD) {508 if (buffer.length >= FLUSH_BUFFER_THRESHOLD) {
479 flushUnsafe();509 Queue.offerUnsafe(flushRequests, undefined);
480 }510 }
481 } catch {511 } catch {
482 return;512 return;