| author |  Theo Browne <me@t3.gg> 2026-10-05 21:10:00-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-05 21:10:00-07:00 |
| log | 570b72d83978f718a5e677002bde3abe55a6d797 |
| tree | ddf9843e74670e1a49a25772348d42cb608b9af7 |
| parent | b69f8b5ff079cccc259bf2a001c2da2bc25d6892 |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(observability): a failing trace disk no longer stalls the server (#13758)
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>3 files changed, 132 insertions(+), 7 deletions(-)
docs
operations
packages
docs/operations/observability.md+1-1
| ... | @@ -579,7 +579,7 @@ Local trace file: | ... | @@ -579,7 +579,7 @@ Local trace file: |
| 579 | - `T3CODE_TRACE_FILE`: override trace file path | 579 | - `T3CODE_TRACE_FILE`: override trace file path |
| 580 | - `T3CODE_TRACE_MAX_BYTES`: per-file rotation size, default `10485760` | 580 | - `T3CODE_TRACE_MAX_BYTES`: per-file rotation size, default `10485760` |
| 581 | - `T3CODE_TRACE_MAX_FILES`: rotated file count, default `10` | 581 | - `T3CODE_TRACE_MAX_FILES`: rotated file count, default `10` |
| 582 | - `T3CODE_TRACE_BATCH_WINDOW_MS`: flush window, default `200` | 582 | - `T3CODE_TRACE_BATCH_WINDOW_MS`: flush window, default `1000` |
| 583 | - `T3CODE_TRACE_MIN_LEVEL`: minimum trace level, default `Info` | 583 | - `T3CODE_TRACE_MIN_LEVEL`: minimum trace level, default `Info` |
| 584 | - `T3CODE_TRACE_TIMING_ENABLED`: enable timing metadata, default `true` | 584 | - `T3CODE_TRACE_TIMING_ENABLED`: enable timing metadata, default `true` |
| 585 | 585 |
packages/shared/src/observability.test.ts+95
| ... | @@ -13,7 +13,10 @@ import * as Ref from "effect/Ref"; | ... | @@ -13,7 +13,10 @@ import * as Ref from "effect/Ref"; |
| 13 | import * as References from "effect/References"; | 13 | import * as References from "effect/References"; |
| 14 | import * as Schema from "effect/Schema"; | 14 | import * as Schema from "effect/Schema"; |
| 15 | import * as Tracer from "effect/Tracer"; | 15 | import * as Tracer from "effect/Tracer"; |
| 16 | import * as TestClock from "effect/testing/TestClock"; | ||
| 17 | import { vi } from "vite-plus/test"; | ||
| 16 | 18 | ||
| 19 | import { RotatingFileSink } from "./logging.ts"; | ||
| 17 | import { | 20 | import { |
| 18 | causeErrorTag, | 21 | causeErrorTag, |
| 19 | compactTraceAttributes, | 22 | compactTraceAttributes, |
| ... | @@ -527,6 +530,98 @@ describe("observability", () => { | ... | @@ -527,6 +530,98 @@ describe("observability", () => { |
| 527 | ), | 530 | ), |
| 528 | ); | 531 | ); |
| 529 | 532 | ||
| 533 | it.effect("drops records after a failed write and logs once per failure episode", () => { | ||
| 534 | const logs: Array<{ readonly logLevel: string; readonly message: unknown }> = []; | ||
| 535 | const captureLogs = Logger.make(({ logLevel, message }) => { | ||
| 536 | logs.push({ logLevel, message }); | ||
| 537 | }); | ||
| 538 | const spans: Array<Tracer.NativeSpan> = []; | ||
| 539 | const recordingTracer = Tracer.make({ | ||
| 540 | span: (options) => { | ||
| 541 | const span = new Tracer.NativeSpan(options); | ||
| 542 | spans.push(span); | ||
| 543 | return span; | ||
| 544 | }, | ||
| 545 | }); | ||
| 546 | |||
| 547 | return Effect.gen(function* () { | ||
| 548 | const fileSystem = yield* FileSystem.FileSystem; | ||
| 549 | const path = yield* Path.Path; | ||
| 550 | const tempDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-trace-sink-" }); | ||
| 551 | const tracePath = path.join(tempDir, "shared.trace.ndjson"); | ||
| 552 | // A directory at the trace path fails every append, like a full disk. | ||
| 553 | yield* fileSystem.makeDirectory(tracePath); | ||
| 554 | const write = vi.spyOn(RotatingFileSink.prototype, "write"); | ||
| 555 | yield* Effect.addFinalizer(() => Effect.sync(() => write.mockRestore())); | ||
| 556 | |||
| 557 | const sink = yield* makeTraceSink({ | ||
| 558 | filePath: tracePath, | ||
| 559 | maxBytes: 1024 * 1024, | ||
| 560 | maxFiles: 2, | ||
| 561 | batchWindowMs: 1_000, | ||
| 562 | }).pipe(Effect.withTracer(recordingTracer)); | ||
| 563 | |||
| 564 | for (let index = 0; index < 1_024; index += 1) { | ||
| 565 | sink.push(makeRecord("lost", String(index))); | ||
| 566 | } | ||
| 567 | // Timed flushes run in the fiber forked inside the makeTraceSink span. | ||
| 568 | for (let index = 0; index < 5; index += 1) { | ||
| 569 | sink.push(makeRecord("lost")); | ||
| 570 | yield* TestClock.adjust("1 second"); | ||
| 571 | } | ||
| 572 | |||
| 573 | // One write per batch, never a growing backlog. | ||
| 574 | assert.deepStrictEqual( | ||
| 575 | write.mock.calls.map(([chunk]) => String(chunk).split("\n").length - 1), | ||
| 576 | [256, 256, 256, 256, 1, 1, 1, 1, 1], | ||
| 577 | ); | ||
| 578 | expect(logs).toEqual([ | ||
| 579 | { logLevel: "Warn", message: [expect.any(String), { filePath: tracePath }] }, | ||
| 580 | ]); | ||
| 581 | |||
| 582 | // Once the disk recovers, new records are written and the loss is reported. | ||
| 583 | yield* fileSystem.remove(tracePath, { recursive: true }); | ||
| 584 | sink.push(makeRecord("recovered")); | ||
| 585 | yield* TestClock.adjust("1 second"); | ||
| 586 | |||
| 587 | const records = yield* readTraceRecords(tracePath); | ||
| 588 | assert.deepStrictEqual( | ||
| 589 | records.map((record) => record.name), | ||
| 590 | ["recovered"], | ||
| 591 | ); | ||
| 592 | expect(logs[1]).toEqual({ | ||
| 593 | logLevel: "Info", | ||
| 594 | message: [expect.any(String), { filePath: tracePath, droppedCount: 1_029 }], | ||
| 595 | }); | ||
| 596 | |||
| 597 | // Healthy flushes after the recovery log nothing. | ||
| 598 | sink.push(makeRecord("healthy")); | ||
| 599 | yield* TestClock.adjust("1 second"); | ||
| 600 | expect(logs).toHaveLength(2); | ||
| 601 | |||
| 602 | // A new failure episode warns again. | ||
| 603 | yield* fileSystem.remove(tracePath); | ||
| 604 | yield* fileSystem.makeDirectory(tracePath); | ||
| 605 | sink.push(makeRecord("lost-again")); | ||
| 606 | yield* TestClock.adjust("1 second"); | ||
| 607 | assert.deepStrictEqual( | ||
| 608 | logs.map((log) => log.logLevel), | ||
| 609 | ["Warn", "Info", "Warn"], | ||
| 610 | ); | ||
| 611 | |||
| 612 | // The ended makeTraceSink span is never released, so it must not collect log events. | ||
| 613 | assert.deepStrictEqual( | ||
| 614 | spans.map((span) => [span.name, span.events.length]), | ||
| 615 | [["makeTraceSink", 0]], | ||
| 616 | ); | ||
| 617 | }).pipe( | ||
| 618 | Effect.scoped, | ||
| 619 | Effect.provide( | ||
| 620 | Logger.layer([captureLogs, Logger.tracerLogger], { mergeWithExisting: false }), | ||
| 621 | ), | ||
| 622 | ); | ||
| 623 | }); | ||
| 624 | |||
| 530 | it.effect("writes nested spans to disk and captures log messages as span events", () => | 625 | it.effect("writes nested spans to disk and captures log messages as span events", () => |
| 531 | Effect.scoped( | 626 | Effect.scoped( |
| 532 | Effect.gen(function* () { | 627 | Effect.gen(function* () { |
packages/shared/src/observability.ts+36-6
| ... | @@ -1,4 +1,5 @@ | ... | @@ -1,4 +1,5 @@ |
| 1 | import * as Cause from "effect/Cause"; | 1 | import * as Cause from "effect/Cause"; |
| 2 | import * as Context from "effect/Context"; | ||
| 2 | import * as Effect from "effect/Effect"; | 3 | import * as Effect from "effect/Effect"; |
| 3 | import type * as Exit from "effect/Exit"; | 4 | import type * as Exit from "effect/Exit"; |
| 4 | import * as ExitRuntime from "effect/Exit"; | 5 | import * as ExitRuntime from "effect/Exit"; |
| ... | @@ -371,6 +372,12 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac | ... | @@ -371,6 +372,12 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac |
| 371 | }); | 372 | }); |
| 372 | 373 | ||
| 373 | let buffer: Array<string> = []; | 374 | let buffer: Array<string> = []; |
| 375 | // 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 a | ||
| 377 | // disk that flaps inside one window stays one episode. | ||
| 378 | let writeFailing = false; | ||
| 379 | let droppedCount = 0; | ||
| 380 | let failureReported = false; | ||
| 374 | let pendingFlushStats: TraceSinkFlushStats = { | 381 | let pendingFlushStats: TraceSinkFlushStats = { |
| 375 | logicalWriteBytes: 0, | 382 | logicalWriteBytes: 0, |
| 376 | count: 0, | 383 | count: 0, |
| ... | @@ -407,9 +414,14 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac | ... | @@ -407,9 +414,14 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac |
| 407 | try { | 414 | try { |
| 408 | sink.write(chunk); | 415 | sink.write(chunk); |
| 409 | } catch { | 416 | } catch { |
| 410 | buffer.unshift(...records.slice(persistedCount)); | 417 | // A failing disk (ENOSPC, EACCES, EIO) drops the rest of the batch. |
| 418 | // Retrying it would grow the backlog, and every later push would | ||
| 419 | // retry all of it. | ||
| 420 | writeFailing = true; | ||
| 421 | droppedCount += records.length - persistedCount; | ||
| 411 | return; | 422 | return; |
| 412 | } | 423 | } |
| 424 | writeFailing = false; | ||
| 413 | pendingFlushStats = { | 425 | pendingFlushStats = { |
| 414 | logicalWriteBytes: pendingFlushStats.logicalWriteBytes + chunkBytes, | 426 | logicalWriteBytes: pendingFlushStats.logicalWriteBytes + chunkBytes, |
| 415 | count: pendingFlushStats.count + nextIndex - persistedCount, | 427 | count: pendingFlushStats.count + nextIndex - persistedCount, |
| ... | @@ -419,7 +431,9 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac | ... | @@ -419,7 +431,9 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac |
| 419 | } | 431 | } |
| 420 | }; | 432 | }; |
| 421 | 433 | ||
| 422 | const flush = Effect.sync(() => { | 434 | // 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. | ||
| 436 | const flush = Effect.gen(function* () { | ||
| 423 | flushUnsafe(); | 437 | flushUnsafe(); |
| 424 | const stats = pendingFlushStats; | 438 | const stats = pendingFlushStats; |
| 425 | pendingFlushStats = { | 439 | pendingFlushStats = { |
| ... | @@ -427,12 +441,28 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac | ... | @@ -427,12 +441,28 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac |
| 427 | count: 0, | 441 | count: 0, |
| 428 | durationMs: 0, | 442 | durationMs: 0, |
| 429 | }; | 443 | }; |
| 430 | return stats; | 444 | if (stats.count > 0 && options.onFlush) { |
| 445 | yield* options.onFlush(stats).pipe(Effect.ignore); | ||
| 446 | } | ||
| 447 | if (droppedCount > 0 && !failureReported) { | ||
| 448 | failureReported = true; | ||
| 449 | yield* Effect.logWarning("Trace writes are failing, dropping records until they recover", { | ||
| 450 | filePath: options.filePath, | ||
| 451 | }); | ||
| 452 | } | ||
| 453 | if (failureReported && !writeFailing) { | ||
| 454 | yield* Effect.logInfo("Trace writes recovered", { | ||
| 455 | filePath: options.filePath, | ||
| 456 | droppedCount, | ||
| 457 | }); | ||
| 458 | failureReported = false; | ||
| 459 | droppedCount = 0; | ||
| 460 | } | ||
| 431 | }).pipe( | 461 | }).pipe( |
| 432 | Effect.flatMap((stats) => | ||
| 433 | stats.count > 0 && options.onFlush ? options.onFlush(stats).pipe(Effect.ignore) : Effect.void, | ||
| 434 | ), | ||
| 435 | Effect.withTracerEnabled(false), | 462 | Effect.withTracerEnabled(false), |
| 463 | // The timed fiber inherits the makeTraceSink span. That span has ended but | ||
| 464 | // lives as long as the sink, so the tracer logger must not add logs to it. | ||
| 465 | Effect.updateContext(Context.omit(Tracer.ParentSpan)<never>), | ||
| 436 | ); | 466 | ); |
| 437 | 467 | ||
| 438 | yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore)); | 468 | yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore)); |