Skip to content

Commit 434ee37

Browse files
t3dotggclaude
t3dotgg
and
claude
committed
fix(observability): log trace write failures once per episode
The drop warning ran on every flush while the disk stayed broken, and the tracer logger added each one as an event on the ended makeTraceSink span that the timed flush fiber keeps alive. Now the sink warns once when writes start failing and logs the total dropped once they recover. The flush also runs without the inherited parent span. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
1 parent c0fdab2 commit 434ee37

2 files changed

Lines changed: 79 additions & 22 deletions

File tree

‎packages/shared/src/observability.test.ts‎

Lines changed: 49 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import * as Ref from "effect/Ref";
1212
import * as References from "effect/References";
1313
import * as Schema from "effect/Schema";
1414
import * as Tracer from "effect/Tracer";
15+
import * as TestClock from "effect/testing/TestClock";
1516
import { vi } from "vite-plus/test";
1617

1718
import { RotatingFileSink } from "./logging.ts";
@@ -378,10 +379,18 @@ describe("observability", () => {
378379
),
379380
);
380381

381-
it.effect("drops records after a failed write instead of retrying them on every push", () => {
382-
const warnings: Array<unknown> = [];
383-
const captureWarnings = Logger.make(({ logLevel, message }) => {
384-
if (logLevel === "Warn") warnings.push(message);
382+
it.effect("drops records after a failed write and logs once per failure episode", () => {
383+
const logs: Array<{ readonly logLevel: string; readonly message: unknown }> = [];
384+
const captureLogs = Logger.make(({ logLevel, message }) => {
385+
logs.push({ logLevel, message });
386+
});
387+
const spans: Array<Tracer.NativeSpan> = [];
388+
const recordingTracer = Tracer.make({
389+
span: (options) => {
390+
const span = new Tracer.NativeSpan(options);
391+
spans.push(span);
392+
return span;
393+
},
385394
});
386395

387396
return Effect.gen(function* () {
@@ -398,37 +407,62 @@ describe("observability", () => {
398407
filePath: tracePath,
399408
maxBytes: 1024 * 1024,
400409
maxFiles: 2,
401-
batchWindowMs: 10_000,
402-
});
410+
batchWindowMs: 1_000,
411+
}).pipe(Effect.withTracer(recordingTracer));
403412

404413
for (let index = 0; index < 1_024; index += 1) {
405414
sink.push(makeRecord("lost", String(index)));
406415
}
407-
yield* sink.flush;
416+
// Timed flushes run in the fiber forked inside the makeTraceSink span.
417+
for (let index = 0; index < 5; index += 1) {
418+
sink.push(makeRecord("lost"));
419+
yield* TestClock.adjust("1 second");
420+
}
408421

409-
// One write per full batch of 256, never a growing backlog.
422+
// One write per batch, never a growing backlog.
410423
assert.deepStrictEqual(
411424
write.mock.calls.map(([chunk]) => String(chunk).split("\n").length - 1),
412-
[256, 256, 256, 256],
425+
[256, 256, 256, 256, 1, 1, 1, 1, 1],
413426
);
414-
expect(warnings).toEqual([
415-
[expect.any(String), { filePath: tracePath, droppedCount: 1_024 }],
427+
expect(logs).toEqual([
428+
{ logLevel: "Warn", message: [expect.any(String), { filePath: tracePath }] },
416429
]);
417430

418-
// Once the disk recovers, new records are written again.
431+
// Once the disk recovers, new records are written and the loss is reported.
419432
yield* fileSystem.remove(tracePath, { recursive: true });
420433
sink.push(makeRecord("recovered"));
421-
yield* sink.flush;
434+
yield* TestClock.adjust("1 second");
422435

423436
const records = yield* readTraceRecords(tracePath);
424437
assert.deepStrictEqual(
425438
records.map((record) => record.name),
426439
["recovered"],
427440
);
428-
assert.equal(warnings.length, 1);
441+
expect(logs[1]).toEqual({
442+
logLevel: "Info",
443+
message: [expect.any(String), { filePath: tracePath, droppedCount: 1_029 }],
444+
});
445+
446+
// A new failure episode warns again.
447+
yield* fileSystem.remove(tracePath);
448+
yield* fileSystem.makeDirectory(tracePath);
449+
sink.push(makeRecord("lost-again"));
450+
yield* TestClock.adjust("1 second");
451+
assert.deepStrictEqual(
452+
logs.map((log) => log.logLevel),
453+
["Warn", "Info", "Warn"],
454+
);
455+
456+
// The ended makeTraceSink span is never released, so it must not collect log events.
457+
assert.deepStrictEqual(
458+
spans.map((span) => [span.name, span.events.length]),
459+
[["makeTraceSink", 0]],
460+
);
429461
}).pipe(
430462
Effect.scoped,
431-
Effect.provide(Logger.layer([captureWarnings], { mergeWithExisting: false })),
463+
Effect.provide(
464+
Logger.layer([captureLogs, Logger.tracerLogger], { mergeWithExisting: false }),
465+
),
432466
);
433467
});
434468

‎packages/shared/src/observability.ts‎

Lines changed: 30 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import * as Cause from "effect/Cause";
2+
import * as Context from "effect/Context";
23
import * as Effect from "effect/Effect";
34
import type * as Exit from "effect/Exit";
45
import * as ExitRuntime from "effect/Exit";
@@ -371,8 +372,12 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac
371372
});
372373

373374
let buffer: Array<string> = [];
374-
// Records lost to failed writes since the last flush reported them.
375+
// Failure episode state. The latest write result says if the disk is
376+
// failing. Records lost since the episode started are counted until a write
377+
// succeeds and the flush reports the recovery.
378+
let writeFailing = false;
375379
let droppedCount = 0;
380+
let failureReported = false;
376381
let pendingFlushStats: TraceSinkFlushStats = {
377382
logicalWriteBytes: 0,
378383
count: 0,
@@ -412,9 +417,11 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac
412417
// A failing disk (ENOSPC, EACCES, EIO) drops the rest of the batch.
413418
// Retrying it would grow the backlog, and every later push would
414419
// retry all of it.
420+
writeFailing = true;
415421
droppedCount += records.length - persistedCount;
416422
return;
417423
}
424+
writeFailing = false;
418425
pendingFlushStats = {
419426
logicalWriteBytes: pendingFlushStats.logicalWriteBytes + chunkBytes,
420427
count: pendingFlushStats.count + nextIndex - persistedCount,
@@ -424,26 +431,42 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac
424431
}
425432
};
426433

434+
// Logs only when writes start failing and when they recover, so a disk that
435+
// stays broken costs one line, not one per flush. It drops the parent span:
436+
// the timed fiber inherits the makeTraceSink span, which has ended but stays
437+
// referenced for the life of the sink, and the tracer logger would add every
438+
// log to it as an event.
427439
const flush = Effect.gen(function* () {
428440
flushUnsafe();
429441
const stats = pendingFlushStats;
430-
const dropped = droppedCount;
431442
pendingFlushStats = {
432443
logicalWriteBytes: 0,
433444
count: 0,
434445
durationMs: 0,
435446
};
436-
droppedCount = 0;
437447
if (stats.count > 0 && options.onFlush) {
438448
yield* options.onFlush(stats).pipe(Effect.ignore);
439449
}
440-
if (dropped > 0) {
441-
yield* Effect.logWarning("Dropped trace records after a failed write", {
450+
if (droppedCount > 0 && !failureReported) {
451+
failureReported = true;
452+
yield* Effect.logWarning("Trace writes are failing, dropping records until they recover", {
442453
filePath: options.filePath,
443-
droppedCount: dropped,
444454
});
445455
}
446-
}).pipe(Effect.withTracerEnabled(false));
456+
if (failureReported && !writeFailing) {
457+
yield* Effect.logInfo("Trace writes recovered", {
458+
filePath: options.filePath,
459+
droppedCount,
460+
});
461+
failureReported = false;
462+
droppedCount = 0;
463+
}
464+
}).pipe(
465+
Effect.withTracerEnabled(false),
466+
Effect.updateContext((context: Context.Context<never>) =>
467+
Context.omit(Tracer.ParentSpan)(context),
468+
),
469+
);
447470

448471
yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore));
449472
yield* Effect.forkScoped(

0 commit comments

Comments
 (0)