From 4befc863f54a3341911c346c8822286ec37e54af Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 05:37:03 -0700 Subject: [PATCH] perf(server): traces keep ~40% more history and no longer block the event loop (#19430) Co-authored-by: Aaron Queen Co-authored-by: Daniyal <115576848+daniyal-sh@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 7 +- .../FoundationPersistence.test.ts | 26 +++ .../src/provider/EventNdjsonLogger.test.ts | 120 +++++++++--- apps/server/src/provider/EventNdjsonLogger.ts | 182 ++++++++++-------- apps/server/src/serverRuntimeStartup.ts | 24 ++- docs/internals/performance-regressions.md | 6 + docs/operations/observability.md | 3 +- packages/shared/src/logging.test.ts | 61 ++++++ packages/shared/src/logging.ts | 83 ++++++++ packages/shared/src/observability.test.ts | 76 +++++++- packages/shared/src/observability.ts | 136 ++++++++----- 11 files changed, 550 insertions(+), 174 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 9ff7a3b24966b5c8d0f50e82fbf17cdcd4774a7d..d591c0245b7effab3227a2e48cd93ec01a604861 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -399,7 +399,12 @@ const layerBase: Layer.Layer< yield* applyStoredEvents(committed); yield* effectOutbox.enqueue(input.effects); return committed; - }), + }).pipe( + // Every streamed agent event is a write, so its statements would + // dominate the trace file. This span and its sql.transaction span + // still time each write, including the commit. + Effect.withTracerEnabled(false), + ), (storedEvents) => Effect.gen(function* () { if (input.effects.length > 0) { diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 023b1dbf22563ccb80051b62073ee67be7698304..e3082f5382dea0105d49bba98b347e62319545b2 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -447,6 +447,32 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { ), ); + it.effect("traces an event write and its transaction without its SQL statements", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const thread = makeThread(ThreadId.make("thread:untraced-write-statements"), now); + const names: Array = []; + const tracer = Tracer.make({ + span: (options) => { + names.push(options.name); + return new Tracer.NativeSpan(options); + }, + }); + + yield* sink + .write({ + events: [ + threadCreatedEvent({ id: "event:untraced-write-statements:created", thread, now }), + ], + }) + .pipe(Effect.withTracer(tracer)); + + // These two spans time a slow commit; the statements inside are not traced. + assert.deepEqual(names, ["orchestrationV2.EventSink.write", "sql.transaction"]); + }), + ); + it.effect.each(["high-water", "replay"] as const)( "bounds live events while the V2 %s query is blocked", (phase) => diff --git a/apps/server/src/provider/EventNdjsonLogger.test.ts b/apps/server/src/provider/EventNdjsonLogger.test.ts index f150596f65ece0f958d224b756ec87470625cf17..256e26e285304d8385c6d51812d4e34746dd8805 100644 --- a/apps/server/src/provider/EventNdjsonLogger.test.ts +++ b/apps/server/src/provider/EventNdjsonLogger.test.ts @@ -4,9 +4,13 @@ import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import { ThreadId } from "@t3tools/contracts"; -import { assert, describe, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { RotatingFileSinkError } from "@t3tools/shared/logging"; import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as FileSystem from "effect/FileSystem"; import * as Fiber from "effect/Fiber"; import * as Logger from "effect/Logger"; import * as Schema from "effect/Schema"; @@ -49,7 +53,7 @@ function parseLogLine(line: string) { }; } -describe("EventNdjsonLogger", () => { +it.layer(NodeServices.layer)("EventNdjsonLogger", (it) => { it.effect("summarizes circular events without exposing their contents in diagnostics", () => { const messages: Array = []; const logCapture = Logger.make(({ message }) => { @@ -256,8 +260,8 @@ describe("EventNdjsonLogger", () => { assert.equal(NodeFS.existsSync(threadPath), false); yield* TestClock.adjust(1_000); - assert.equal(NodeFS.existsSync(threadPath), true); yield* store.close(); + assert.equal(NodeFS.existsSync(threadPath), true); } finally { NodeFS.rmSync(tempDir, { recursive: true, force: true }); } @@ -281,10 +285,10 @@ describe("EventNdjsonLogger", () => { yield* logger.write({ id: "accepted" }, ThreadId.make("thread-interrupted")); yield* TestClock.adjust(1_000); + yield* store.close(); assert.equal(NodeFS.existsSync(threadPath), true); assert.include(NodeFS.readFileSync(threadPath, "utf8"), '{"id":"accepted"}'); - yield* store.close(); } finally { NodeFS.rmSync(tempDir, { recursive: true, force: true }); } @@ -788,39 +792,99 @@ describe("EventNdjsonLogger", () => { }), ); - it("attributes batches that were written before a later chunk fails", () => { - const records: ReadonlyArray = [ - { - stream: "native", - threadSegment: "thread", - line: "first", - bytes: 5, - }, - { - stream: "canonical", - threadSegment: "thread", - line: "second", - bytes: 6, - }, - ]; - const attributed: Array = []; - let writes = 0; + it.effect("attributes batches that were written before a later chunk fails", () => + Effect.gen(function* () { + const records: ReadonlyArray = [ + { + stream: "native", + threadSegment: "thread", + line: "first", + bytes: 5, + }, + { + stream: "canonical", + threadSegment: "thread", + line: "second", + bytes: 6, + }, + ]; + const attributed: Array = []; + let writes = 0; - assert.throws(() => - writeBatchedMessages( + const result = yield* writeBatchedMessages( { write: () => { writes += 1; - if (writes === 2) throw new Error("simulated disk exhaustion"); + return writes === 2 + ? Effect.fail( + new RotatingFileSinkError({ + operation: "write", + filePath: "test.log", + cause: "simulated disk exhaustion", + }), + ) + : Effect.void; }, }, records, 5, (written) => attributed.push(...written), - ), - ); - assert.deepEqual(attributed, [records[0]]); - }); + ).pipe(Effect.result); + assert.equal(result._tag, "Failure"); + assert.deepEqual(attributed, [records[0]]); + }), + ); + + it.effect( + "keeps unrelated work responsive and drains before close while a disk write is held", + () => + Effect.gen(function* () { + const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-")); + const basePath = NodePath.join(tempDir, "events.log"); + const fs = yield* FileSystem.FileSystem; + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + let writes = 0; + try { + const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 }).pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + writeFile: (filePath, content, options) => + Effect.gen(function* () { + if (++writes === 1) { + yield* Deferred.succeed(started, undefined); + yield* Deferred.await(release); + } + yield* fs.writeFile(filePath, content, options); + }), + }), + ); + const first = yield* store + .logger("native") + .write({ id: "first" }, ThreadId.make("held")) + .pipe(Effect.forkChild); + yield* Deferred.await(started); + const closed = yield* Deferred.make(); + const close = yield* store + .close() + .pipe(Effect.andThen(Deferred.succeed(closed, undefined)), Effect.forkChild); + assert.equal(yield* Deferred.isDone(closed), false); + // A separate fiber runs while the file operation remains pending. + const heartbeat = yield* Deferred.make(); + yield* Effect.forkChild(Deferred.succeed(heartbeat, undefined)); + yield* Deferred.await(heartbeat); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(first); + yield* Fiber.join(close); + assert.include( + NodeFS.readFileSync(ownedLogPath(basePath, "held"), "utf8"), + '{"id":"first"}', + ); + } finally { + NodeFS.rmSync(tempDir, { recursive: true, force: true }); + } + }), + ); it.effect("reports logical provider log writes to resource attribution", () => Effect.gen(function* () { diff --git a/apps/server/src/provider/EventNdjsonLogger.ts b/apps/server/src/provider/EventNdjsonLogger.ts index f8cbca0c1316a6cd5460dea7cfed241ba28152a1..7e5a133209b80fa0902b0107ea98462f82c261c4 100644 --- a/apps/server/src/provider/EventNdjsonLogger.ts +++ b/apps/server/src/provider/EventNdjsonLogger.ts @@ -10,11 +10,14 @@ import * as NodePath from "node:path"; import type { ThreadId } from "@t3tools/contracts"; import type * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers"; -import { RotatingFileSink } from "@t3tools/shared/logging"; +import { makeAsyncRotatingFileSink } from "@t3tools/shared/logging"; import { errorTag } from "@t3tools/shared/observability"; import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Result from "effect/Result"; import * as Exit from "effect/Exit"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; @@ -135,10 +138,12 @@ export interface PendingRecord { readonly bytes: number; } +type AsyncFileSink = Effect.Success>; + interface StoreState { readonly pending: Array; readonly pendingBytes: number; - readonly sinks: ReadonlyMap; + readonly sinks: ReadonlyMap; readonly flushScheduled: boolean; readonly closed: boolean; readonly lastRetentionAt: number; @@ -369,65 +374,69 @@ export function boundProviderEventForLogging(event: unknown): unknown { return summarizeProviderEvent(event); } -export function writeBatchedMessages( - sink: Pick, +export const writeBatchedMessages = Effect.fnUntraced(function* ( + sink: AsyncFileSink, records: ReadonlyArray, maxBytes: number, onWritten: (records: ReadonlyArray) => void, -): void { +) { let pendingRecords: Array = []; let pendingBytes = 0; - const flush = () => { + const flush = Effect.gen(function* () { if (pendingRecords.length === 0) return; const writtenRecords = pendingRecords; - sink.write(writtenRecords.map((record) => record.line).join("")); + yield* sink.write(writtenRecords.map((record) => record.line).join("")); onWritten(writtenRecords); pendingRecords = []; pendingBytes = 0; - }; + }); for (const record of records) { if (pendingBytes > 0 && pendingBytes + record.bytes > maxBytes) { - flush(); + yield* flush; } pendingRecords.push(record); pendingBytes += record.bytes; if (pendingBytes >= maxBytes) { - flush(); + yield* flush; } } - flush(); -} + yield* flush; +}); -function isProviderLogFile(filePath: string, fileName: string, filePrefix: string): boolean { +async function isProviderLogFile( + filePath: string, + fileName: string, + filePrefix: string, +): Promise { if (!/\.log(?:\.\d+)?$/u.test(fileName)) return false; if (fileName.startsWith(filePrefix)) return true; - const descriptor = NodeFS.openSync(filePath, "r"); + const descriptor = await NodeFS.promises.open(filePath, "r"); try { const header = Buffer.alloc(256); - const bytesRead = NodeFS.readSync(descriptor, header, 0, header.byteLength, 0); + const { bytesRead } = await descriptor.read(header, 0, header.byteLength, 0); return /^\[[^\]\r\n]+\] (?:NTIVE|CANON|ORCH): /u.test(header.toString("utf8", 0, bytesRead)); } finally { - NodeFS.closeSync(descriptor); + await descriptor.close(); } } -function enforceRetention(input: { +async function enforceRetention(input: { readonly directory: string; readonly maxTotalBytes: number; readonly maxAgeMs: number; readonly activeFilePaths: ReadonlySet; readonly filePrefix: string; readonly now: number; -}): RetentionResult { +}): Promise { const failures: Array = []; const files: Array<{ filePath: string; mtimeMs: number; size: number }> = []; let entries: ReadonlyArray; try { - entries = NodeFS.readdirSync(input.directory, { withFileTypes: true }); + entries = await NodeFS.promises.readdir(input.directory, { withFileTypes: true }); } catch (cause) { return { failures: [{ filePath: input.directory, cause }] }; } @@ -436,8 +445,8 @@ function enforceRetention(input: { if (!entry.isFile()) continue; const filePath = NodePath.join(input.directory, entry.name); try { - if (!isProviderLogFile(filePath, entry.name, input.filePrefix)) continue; - const stat = NodeFS.statSync(filePath); + if (!(await isProviderLogFile(filePath, entry.name, input.filePrefix))) continue; + const stat = await NodeFS.promises.stat(filePath); files.push({ filePath, mtimeMs: stat.mtimeMs, size: stat.size }); } catch (cause) { failures.push({ filePath, cause }); @@ -445,10 +454,10 @@ function enforceRetention(input: { } let totalBytes = files.reduce((total, file) => total + file.size, 0); - const remove = (file: (typeof files)[number]) => { + const remove = async (file: (typeof files)[number]) => { if (input.activeFilePaths.has(file.filePath)) return false; try { - NodeFS.rmSync(file.filePath, { force: true }); + await NodeFS.promises.rm(file.filePath, { force: true }); totalBytes -= file.size; return true; } catch (cause) { @@ -457,16 +466,16 @@ function enforceRetention(input: { } }; - const retained = files.filter((file) => { - if (input.now - file.mtimeMs <= input.maxAgeMs) return true; - return !remove(file); - }); + const retained: typeof files = []; + for (const file of files) { + if (input.now - file.mtimeMs <= input.maxAgeMs || !(await remove(file))) retained.push(file); + } for (const file of retained.toSorted( (left, right) => left.mtimeMs - right.mtimeMs || left.filePath.localeCompare(right.filePath), )) { if (totalBytes <= input.maxTotalBytes) break; - remove(file); + await remove(file); } return { failures }; @@ -517,7 +526,7 @@ function resolveOptions( return Effect.succeed(resolved); } -function drainPending(input: { +const drainPending = Effect.fnUntraced(function* (input: { readonly directory: string; readonly options: ResolvedOptions; readonly state: StoreState; @@ -525,7 +534,10 @@ function drainPending(input: { readonly now: number; readonly timerFired: boolean; readonly close: boolean; -}): readonly [DrainResult, StoreState] { + readonly makeSink: ( + filePath: string, + ) => Effect.Effect>>; +}): Effect.fn.Return { if (input.state.closed) { return [{ attributions: [], failures: [] }, input.state]; } @@ -548,22 +560,20 @@ function drainPending(input: { const filePath = providerLogPath(input.directory, input.filePrefix, threadSegment); let sink = sinks.get(threadSegment); if (!sink) { - try { - sink = new RotatingFileSink({ - filePath, - maxBytes: input.options.maxBytes, - maxFiles: input.options.maxFiles, - throwOnError: true, - }); - sinks.set(threadSegment, sink); - } catch (cause) { - failures.push({ filePath, cause }); + const created = yield* input.makeSink(filePath).pipe(Effect.result); + if (Result.isFailure(created)) { + failures.push({ filePath, cause: created.failure }); continue; } + sink = created.success; + sinks.set(threadSegment, sink); } - try { - writeBatchedMessages(sink, records, input.options.maxBytes, (writtenRecords) => { + const written = yield* writeBatchedMessages( + sink, + records, + input.options.maxBytes, + (writtenRecords) => { for (const record of writtenRecords) { const current = attributionByStream.get(record.stream) ?? { count: 0, @@ -574,28 +584,31 @@ function drainPending(input: { logicalWriteBytes: current.logicalWriteBytes + record.bytes, }); } - }); - } catch (cause) { + }, + ).pipe(Effect.result); + if (Result.isFailure(written)) { sinks.delete(threadSegment); - failures.push({ filePath, cause }); + failures.push({ filePath, cause: written.failure }); } } const retentionDue = input.now - input.state.lastRetentionAt >= input.options.retentionCheckIntervalMs; const retention = retentionDue - ? enforceRetention({ - directory: input.directory, - maxTotalBytes: input.options.maxTotalBytes, - maxAgeMs: input.options.maxAgeMs, - activeFilePaths: new Set( - Array.from(sinks.keys(), (threadSegment) => - providerLogPath(input.directory, input.filePrefix, threadSegment), + ? yield* Effect.promise(() => + enforceRetention({ + directory: input.directory, + maxTotalBytes: input.options.maxTotalBytes, + maxAgeMs: input.options.maxAgeMs, + activeFilePaths: new Set( + Array.from(sinks.keys(), (threadSegment) => + providerLogPath(input.directory, input.filePrefix, threadSegment), + ), ), - ), - filePrefix: input.filePrefix, - now: input.now, - }) + filePrefix: input.filePrefix, + now: input.now, + }), + ) : { failures: [] }; return [ @@ -615,7 +628,7 @@ function drainPending(input: { lastRetentionAt: retentionDue ? input.now : input.state.lastRetentionAt, }, ]; -} +}); const serializeEvent = Effect.fnUntraced(function* (event: unknown) { return yield* encodeUnknownJsonString(event).pipe( @@ -630,18 +643,32 @@ const serializeEvent = Effect.fnUntraced(function* (event: unknown) { export const makeEventNdjsonLogStore = Effect.fnUntraced(function* ( filePath: string, options: EventNdjsonLogStoreOptions = {}, -): Effect.fn.Return { +): Effect.fn.Return< + EventNdjsonLogStore, + EventNdjsonLogStoreError, + FileSystem.FileSystem | Path.Path +> { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const resolved = yield* resolveOptions(filePath, options); const directory = NodePath.dirname(filePath); const filePrefix = providerLogPrefix(filePath); - yield* Effect.try({ - try: () => NodeFS.mkdirSync(directory, { recursive: true }), - catch: (cause) => new EventNdjsonLogDirectoryError({ directory, cause }), - }); + yield* fs + .makeDirectory(directory, { recursive: true }) + .pipe(Effect.mapError((cause) => new EventNdjsonLogDirectoryError({ directory, cause }))); + const makeSink = (filePath: string) => + makeAsyncRotatingFileSink({ + filePath, + maxBytes: resolved.maxBytes, + maxFiles: resolved.maxFiles, + }).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + ); const initializedAt = yield* Clock.currentTimeMillis; - const initialRetention = yield* Effect.sync(() => + const initialRetention = yield* Effect.promise(() => enforceRetention({ directory, maxTotalBytes: resolved.maxTotalBytes, @@ -671,17 +698,16 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* ( const flush = Effect.fnUntraced(function* (timerFired: boolean, close: boolean) { const startedAt = yield* Clock.currentTimeMillis; const result = yield* SynchronizedRef.modifyEffect(stateRef, (state) => - Effect.sync(() => - drainPending({ - directory, - options: resolved, - state, - filePrefix, - now: startedAt, - timerFired, - close, - }), - ), + drainPending({ + directory, + options: resolved, + state, + filePrefix, + now: startedAt, + timerFired, + close, + makeSink, + }), ); for (const failure of result.failures) { @@ -714,7 +740,7 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* ( { discard: true }, ); } - }); + }, Effect.uninterruptible); const scheduleFlush = Effect.fnUntraced(function* () { yield* Effect.forkIn( @@ -791,7 +817,11 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* ( export const makeEventNdjsonLogger = Effect.fnUntraced(function* ( filePath: string, options: EventNdjsonLoggerOptions, -): Effect.fn.Return { +): Effect.fn.Return< + ProviderEventLoggers.EventNdjsonLogger | undefined, + never, + FileSystem.FileSystem | Path.Path +> { const store = yield* makeEventNdjsonLogStore(filePath, options).pipe( Effect.catch((error) => logWarning(error.message, { error }).pipe( diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index db17bcecde649628d7b463e0353c0eb868476c64..84c002e3d60d562e731e1ce45ae48ba3cf93bf3c 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -346,8 +346,10 @@ const maybeOpenBrowser = (target: string) => const runStartupPhase = (phase: string, effect: Effect.Effect) => effect.pipe( - Effect.annotateSpans({ "startup.phase": phase }), - Effect.withSpan(`server.startup.${phase}`), + // Attributes on the span itself, not annotateSpans: an annotation follows + // every fiber forked inside the phase, such as the effect worker, for the + // life of the server. + Effect.withSpan(`server.startup.${phase}`, { attributes: { "startup.phase": phase } }), ); interface StartupOptions { @@ -581,8 +583,9 @@ const make = (options?: StartupOptions) => Effect.gen(function* () { yield* Effect.logDebug("startup phase: recording startup heartbeat"); yield* recordStartupHeartbeat.pipe( - Effect.annotateSpans({ "startup.phase": "heartbeat.record" }), - Effect.withSpan("server.startup.heartbeat.record"), + Effect.withSpan("server.startup.heartbeat.record", { + attributes: { "startup.phase": "heartbeat.record" }, + }), Effect.ignoreCause({ log: true }), ); if (serverConfig.startupPresentation === "headless") { @@ -653,12 +656,15 @@ const make = (options?: StartupOptions) => yield* Effect.logDebug("startup phase: complete"); yield* flushCompileCache; }).pipe( - Effect.annotateSpans({ - "server.mode": serverConfig.mode, - "server.port": serverConfig.port, - "server.host": serverConfig.host ?? "default", + Effect.withSpan("server.startup", { + kind: "server", + root: true, + attributes: { + "server.mode": serverConfig.mode, + "server.port": serverConfig.port, + "server.host": serverConfig.host ?? "default", + }, }), - Effect.withSpan("server.startup", { kind: "server", root: true }), ); yield* Effect.forkScoped( diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 9440d2ab6144832a3016fa240d27b29affe3e45e..042fe07b44c811a615674759b2ebf6a4229d2a41 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -90,6 +90,12 @@ counts keep the full status path. Cached fetch failures share the existing backo the actual failed attempt. Unchanged desktop environment bootstrap reads retain their array identity so polling does not invalidate subscribers. +Local trace records and provider event logs use asynchronous, serialized file writes and rotation. +Batching synchronous appends still stalls the server event loop on slow storage, especially on +Windows. While a trace write is slow, producers keep up to one trace file of records, so the spans +that time a stall survive it; records drop only while the disk stays stuck. Provider logging +retains backpressure. Shutdown drains accepted writes before closing their scope. + Relay awareness queues one pending publication per thread and retains a follow-up when the thread changes during a send. Transcript deltas and tool output do not enqueue awareness work. Run, request, and relevant thread metadata events do. Configuration is checked before shell reads, and diff --git a/docs/operations/observability.md b/docs/operations/observability.md index 632bbd34e9f16fd36058849d8f35ea612aee74b3..b434eb6e396cfc3092f38118e5242c7114df9b02 100644 --- a/docs/operations/observability.md +++ b/docs/operations/observability.md @@ -646,7 +646,8 @@ Current high-value span and metric boundaries include: - provider session and turn operations - git command execution and git hook events - terminal session lifecycle -- sqlite query execution +- sqlite query execution, except the statements inside `orchestrationV2.EventSink.write`, which + its `sql.transaction` span times as one - event loop stalls (`server.eventLoop.stall`) ### Current Constraints diff --git a/packages/shared/src/logging.test.ts b/packages/shared/src/logging.test.ts index 7317cb4f7b49f5f3831d82c7ffe7848a8f964306..fdd14c2d75cabdad35fee5069ba420c34be7af23 100644 --- a/packages/shared/src/logging.test.ts +++ b/packages/shared/src/logging.test.ts @@ -3,12 +3,17 @@ import * as NodeFS from "node:fs"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import { afterEach, describe, expect, it } from "vite-plus/test"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { it as effectIt } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as HostProcess from "./HostProcess.ts"; import { RotatingFileSink, RotatingFileSinkConfigurationError, RotatingFileSinkError, + makeAsyncRotatingFileSink, } from "./logging.ts"; const windowsHost = HostProcess.Platform.defaultValue() === "win32"; @@ -35,6 +40,62 @@ afterEach(() => { } }); +effectIt.layer(NodeServices.layer)("asynchronous rotating file sink", (it) => { + it.effect("serializes concurrent appends across rotation and resumes an existing file", () => + Effect.gen(function* () { + const directory = makeTempDirectory(); + const filePath = NodePath.join(directory, "async.log"); + NodeFS.writeFileSync(`${filePath}.50`, "obsolete"); + const lines = Array.from( + { length: 100 }, + (_, index) => `${index.toString().padStart(3, "0")}\n`, + ); + const sink = yield* makeAsyncRotatingFileSink({ filePath, maxBytes: 32, maxFiles: 20 }); + yield* Effect.forEach(lines, (line) => sink.write(line), { + concurrency: "unbounded", + discard: true, + }); + const resumed = yield* makeAsyncRotatingFileSink({ filePath, maxBytes: 32, maxFiles: 20 }); + yield* resumed.write("100\n"); + expect(NodeFS.existsSync(`${filePath}.50`)).toBe(false); + const backups = NodeFS.readdirSync(directory) + .filter((name) => name.startsWith("async.log.")) + .sort((a, b) => Number(b.slice(10)) - Number(a.slice(10))); + const contents = [...backups, "async.log"].map((name) => + NodeFS.readFileSync(NodePath.join(directory, name), "utf8"), + ); + expect(contents.every((chunk) => Buffer.byteLength(chunk) <= 32)).toBe(true); + expect(contents.join("")).toBe(lines.join("") + "100\n"); + }), + ); + + it.effect("reports a failed append and recovers its size before the next rotation", () => + Effect.gen(function* () { + const directory = makeTempDirectory(); + const filePath = NodePath.join(directory, "recovery.log"); + NodeFS.writeFileSync(filePath, "old\n"); + const fs = yield* FileSystem.FileSystem; + let attempts = 0; + const sink = yield* makeAsyncRotatingFileSink({ filePath, maxBytes: 8, maxFiles: 2 }).pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + writeFile: (path, bytes, options) => + ++attempts === 1 + ? fs.writeFile(directory, bytes, options) + : fs.writeFile(path, bytes, options), + }), + ); + const failed = yield* sink.write("new\n").pipe(Effect.result); + expect(failed._tag).toBe("Failure"); + if (failed._tag === "Failure") expect(failed.failure.operation).toBe("write"); + yield* sink.write("new\n"); + yield* sink.write("end\n"); + expect(NodeFS.readFileSync(`${filePath}.1`, "utf8")).toBe("old\nnew\n"); + expect(NodeFS.readFileSync(filePath, "utf8")).toBe("end\n"); + }), + ); +}); + describe("RotatingFileSink", () => { it.each([ { option: "maxBytes" as const, maxBytes: 0, maxFiles: 1 }, diff --git a/packages/shared/src/logging.ts b/packages/shared/src/logging.ts index b75c4f6fac33bcc013ce824bb795dc0f914fd615..a0ab1c16fd8ea35aca38d52201969138add37cbc 100644 --- a/packages/shared/src/logging.ts +++ b/packages/shared/src/logging.ts @@ -2,6 +2,10 @@ import * as NodeFS from "node:fs"; import * as NodePath from "node:path"; import * as Schema from "effect/Schema"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Semaphore from "effect/Semaphore"; export interface RotatingFileSinkOptions { readonly filePath: string; @@ -180,3 +184,82 @@ export class RotatingFileSink { return `${this.filePath}.${index}`; } } + +/** Serialized asynchronous writes and rotation without blocking the event loop. */ +export const makeAsyncRotatingFileSink = Effect.fnUntraced(function* ( + options: Omit, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + if (options.maxBytes < 1) + return yield* new RotatingFileSinkConfigurationError({ + option: "maxBytes", + received: options.maxBytes, + minimum: 1, + }); + if (options.maxFiles < 1) + return yield* new RotatingFileSinkConfigurationError({ + option: "maxFiles", + received: options.maxFiles, + minimum: 1, + }); + const failure = (operation: RotatingFileSinkError["operation"]) => (cause: unknown) => + new RotatingFileSinkError({ operation, filePath: options.filePath, cause }); + const directory = path.dirname(options.filePath); + const basename = path.basename(options.filePath); + yield* fs + .makeDirectory(directory, { recursive: true }) + .pipe(Effect.mapError(failure("initialize"))); + yield* Effect.gen(function* () { + for (const entry of yield* fs.readDirectory(directory)) { + if (!entry.startsWith(`${basename}.`)) continue; + const suffix = Number(entry.slice(basename.length + 1)); + if (!Number.isInteger(suffix) || suffix <= options.maxFiles) continue; + yield* fs.remove(path.join(directory, entry), { force: true }); + } + }).pipe(Effect.mapError(failure("prune"))); + const readSize = fs.stat(options.filePath).pipe( + Effect.map((stat) => Number(stat.size)), + Effect.catch((cause) => + cause.reason._tag === "NotFound" ? Effect.succeed(0) : Effect.fail(failure("read")(cause)), + ), + ); + let currentSize = yield* readSize; + const semaphore = yield* Semaphore.make(1); + const rotate = Effect.gen(function* () { + yield* fs.remove(`${options.filePath}.${options.maxFiles}`, { force: true }); + for (let index = options.maxFiles - 1; index >= 1; index--) { + const source = `${options.filePath}.${index}`; + if (yield* fs.exists(source)) yield* fs.rename(source, `${options.filePath}.${index + 1}`); + } + if (yield* fs.exists(options.filePath)) + yield* fs.rename(options.filePath, `${options.filePath}.1`); + currentSize = 0; + }).pipe(Effect.mapError(failure("rotate"))); + const write = (chunk: string | Buffer) => + semaphore + .withPermits(1)( + Effect.gen(function* () { + const buffer = typeof chunk === "string" ? Buffer.from(chunk) : chunk; + if (buffer.length === 0) return; + if (currentSize > 0 && currentSize + buffer.length > options.maxBytes) yield* rotate; + yield* fs + .writeFile(options.filePath, buffer, { flag: "a" }) + .pipe(Effect.mapError(failure("write"))); + currentSize += buffer.length; + }).pipe( + Effect.tapError(() => + readSize.pipe( + Effect.tap((size) => + Effect.sync(() => { + currentSize = size; + }), + ), + Effect.ignore, + ), + ), + ), + ) + .pipe(Effect.uninterruptible); + return { write }; +}); diff --git a/packages/shared/src/observability.test.ts b/packages/shared/src/observability.test.ts index 8959f7787e76d700b714625c7afe8bbb56549f95..ccc2e04d6d2dd9bd9b89083cbb5f31aa919c67da 100644 --- a/packages/shared/src/observability.test.ts +++ b/packages/shared/src/observability.test.ts @@ -3,6 +3,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import * as Arr from "effect/Array"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; @@ -16,7 +18,6 @@ import * as Tracer from "effect/Tracer"; import * as TestClock from "effect/testing/TestClock"; import { vi } from "vite-plus/test"; -import { RotatingFileSink } from "./logging.ts"; import { causeErrorTag, compactTraceAttributes, @@ -242,7 +243,7 @@ describe("observability", () => { sampled ? [{ _tag: "Success" }] : [], ); } - }), + }).pipe(Effect.provide(NodeServices.layer)), ); it.effect("preserves failure and interruption causes in spans and exported traces", () => @@ -290,7 +291,7 @@ describe("observability", () => { } assert.equal(records[0]?.exit._tag, Cause.hasInterrupts(cause) ? "Interrupted" : "Failure"); } - }), + }).pipe(Effect.provide(NodeServices.layer)), ); it("normalizes circular arrays, maps, and sets without recursing forever", () => { @@ -331,6 +332,64 @@ describe("observability", () => { }); nodeServicesIt("node services", (it) => { + it.effect( + "keeps trace producers responsive during a slow write, bounds the backlog, and drains on close", + () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ + prefix: "t3-trace-backpressure-", + }); + const tracePath = path.join(directory, "trace.ndjson"); + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + const closed = yield* Ref.make(false); + const sink = yield* makeTraceSink({ + filePath: tracePath, + maxBytes: 5_000_000, + maxFiles: 2, + batchWindowMs: 10_000, + }).pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + writeFile: (file, data, options) => + Effect.gen(function* () { + yield* Deferred.succeed(started, undefined); + yield* Deferred.await(release); + yield* fs.writeFile(file, data, options); + }), + }), + ); + sink.push(makeRecord("first")); + const flushing = yield* Effect.forkChild(sink.flush); + yield* Deferred.await(started); + // These pushes return while the write is held. The backlog keeps one + // trace file of records: every small one, then large ones until the + // next would not fit. + for (let index = 0; index < 2_000; index++) + sink.push(makeRecord("small", String(index))); + for (let index = 0; index < 3; index++) + sink.push({ ...makeRecord("large"), attributes: { payload: "x".repeat(2_000_000) } }); + const closing = yield* Effect.forkChild( + sink.close().pipe(Effect.tap(() => Ref.set(closed, true))), + ); + yield* Effect.yieldNow; + assert.isFalse(yield* Ref.get(closed)); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(flushing); + yield* Fiber.join(closing); + const records = yield* readTraceRecords(tracePath); + assert.deepStrictEqual( + records.map((record) => record.name), + ["first", ...Array.from({ length: 2_000 }, () => "small"), "large", "large"], + ); + assert.isTrue(yield* Ref.get(closed)); + }), + ), + ); + it.effect("flushes buffered trace records on close", () => Effect.scoped( Effect.gen(function* () { @@ -551,7 +610,7 @@ describe("observability", () => { const tracePath = path.join(tempDir, "shared.trace.ndjson"); // A directory at the trace path fails every append, like a full disk. yield* fileSystem.makeDirectory(tracePath); - const write = vi.spyOn(RotatingFileSink.prototype, "write"); + const write = vi.spyOn(fileSystem, "writeFile"); yield* Effect.addFinalizer(() => Effect.sync(() => write.mockRestore())); const sink = yield* makeTraceSink({ @@ -564,16 +623,18 @@ describe("observability", () => { for (let index = 0; index < 1_024; index += 1) { sink.push(makeRecord("lost", String(index))); } + yield* sink.flush; // Timed flushes run in the fiber forked inside the makeTraceSink span. for (let index = 0; index < 5; index += 1) { sink.push(makeRecord("lost")); yield* TestClock.adjust("1 second"); + yield* sink.flush; } // One write per batch, never a growing backlog. assert.deepStrictEqual( - write.mock.calls.map(([chunk]) => String(chunk).split("\n").length - 1), - [256, 256, 256, 256, 1, 1, 1, 1, 1], + write.mock.calls.map(([, chunk]) => Buffer.from(chunk).toString().split("\n").length - 1), + [1024, 1, 1, 1, 1, 1], ); expect(logs).toEqual([ { logLevel: "Warn", message: [expect.any(String), { filePath: tracePath }] }, @@ -583,6 +644,7 @@ describe("observability", () => { yield* fileSystem.remove(tracePath, { recursive: true }); sink.push(makeRecord("recovered")); yield* TestClock.adjust("1 second"); + yield* sink.flush; const records = yield* readTraceRecords(tracePath); assert.deepStrictEqual( @@ -597,6 +659,7 @@ describe("observability", () => { // Healthy flushes after the recovery log nothing. sink.push(makeRecord("healthy")); yield* TestClock.adjust("1 second"); + yield* sink.flush; expect(logs).toHaveLength(2); // A new failure episode warns again. @@ -604,6 +667,7 @@ describe("observability", () => { yield* fileSystem.makeDirectory(tracePath); sink.push(makeRecord("lost-again")); yield* TestClock.adjust("1 second"); + yield* sink.flush; assert.deepStrictEqual( logs.map((log) => log.logLevel), ["Warn", "Info", "Warn"], diff --git a/packages/shared/src/observability.ts b/packages/shared/src/observability.ts index a8cdb8059f5a0a55ef0c8f09eb36179694da7735..9e632ca524382c67b88371b0e54660eea0862342 100644 --- a/packages/shared/src/observability.ts +++ b/packages/shared/src/observability.ts @@ -4,13 +4,16 @@ import * as Effect from "effect/Effect"; import type * as Exit from "effect/Exit"; import * as ExitRuntime from "effect/Exit"; import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Result from "effect/Result"; +import * as Semaphore from "effect/Semaphore"; import * as Schema from "effect/Schema"; import * as SchemaIssue from "effect/SchemaIssue"; import * as SchemaTransformation from "effect/SchemaTransformation"; import * as Tracer from "effect/Tracer"; import { OtlpResource, OtlpTracer, OtlpSerialization } from "effect/observability"; -import { RotatingFileSink } from "./logging.ts"; +import { makeAsyncRotatingFileSink } from "./logging.ts"; export const OtlpProtocol = Schema.Literals(["http/json", "http/protobuf"]); export type OtlpProtocol = typeof OtlpProtocol.Type; @@ -37,7 +40,6 @@ export const DEFAULT_SIGNAL_EXPORT: SignalExport = { }; const FLUSH_BUFFER_THRESHOLD = 256; -const textEncoder = new TextEncoder(); export type TraceAttributes = Readonly>; @@ -364,14 +366,20 @@ function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord { } export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) { - const sink = new RotatingFileSink({ + const sink = yield* makeAsyncRotatingFileSink({ filePath: options.filePath, maxBytes: options.maxBytes, maxFiles: options.maxFiles, - throwOnError: true, - }); + }).pipe(Effect.orDie); - let buffer: Array = []; + let buffer: Array<{ readonly text: string; readonly bytes: number }> = []; + let bufferBytes = 0; + // While a write is slow, push keeps up to one trace file of records (at least + // 4 MiB), so the spans that time a stall still reach the file. Only a disk + // that stays stuck drops records, instead of growing memory or blocking push. + const maxBufferBytes = Math.max(options.maxBytes, 4 * 1024 * 1024); + const flushRequests = yield* Queue.sliding(1); + const flushSemaphore = yield* Semaphore.make(1); // A failure episode starts at the first dropped record and ends when a flush // sees that the latest write succeeded. Flush checks once per window, so a // disk that flaps inside one window stays one episode. @@ -384,36 +392,35 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac durationMs: 0, }; - const flushUnsafe = () => { + const flushBuffered = Effect.gen(function* () { if (buffer.length === 0) { return; } const records = buffer; buffer = []; + bufferBytes = 0; let persistedCount = 0; while (persistedCount < records.length) { - const firstRecordBytes = textEncoder.encode(records[persistedCount]).byteLength; - if (firstRecordBytes > options.maxBytes) { - persistedCount += 1; - continue; - } - + // push drops any record larger than a file, so the first one always fits. + const firstRecordBytes = records[persistedCount]!.bytes; let nextIndex = persistedCount + 1; let chunkBytes = firstRecordBytes; while (nextIndex < records.length) { - const nextRecordBytes = textEncoder.encode(records[nextIndex]).byteLength; + const nextRecordBytes = records[nextIndex]!.bytes; if (chunkBytes + nextRecordBytes > options.maxBytes) break; chunkBytes += nextRecordBytes; nextIndex += 1; } - const chunk = records.slice(persistedCount, nextIndex).join(""); + const chunk = records + .slice(persistedCount, nextIndex) + .map((record) => record.text) + .join(""); const startedAt = performance.now(); - try { - sink.write(chunk); - } catch { + const written = yield* sink.write(chunk).pipe(Effect.result); + if (Result.isFailure(written)) { // A failing disk (ENOSPC, EACCES, EIO) drops the rest of the batch. // Retrying it would grow the backlog, and every later push would // retry all of it. @@ -429,54 +436,77 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac }; persistedCount = nextIndex; } - }; + }); // Logs once when writes start failing and once when they recover, so a disk // that stays broken costs one line, not one per flush. - const flush = Effect.gen(function* () { - flushUnsafe(); - const stats = pendingFlushStats; - pendingFlushStats = { - logicalWriteBytes: 0, - count: 0, - durationMs: 0, - }; - if (stats.count > 0 && options.onFlush) { - yield* options.onFlush(stats).pipe(Effect.ignore); - } - if (droppedCount > 0 && !failureReported) { - failureReported = true; - yield* Effect.logWarning("Trace writes are failing, dropping records until they recover", { - filePath: options.filePath, - }); - } - if (failureReported && !writeFailing) { - yield* Effect.logInfo("Trace writes recovered", { - filePath: options.filePath, - droppedCount, - }); - failureReported = false; - droppedCount = 0; - } - }).pipe( - Effect.withTracerEnabled(false), - // The timed fiber inherits the makeTraceSink span. That span has ended but - // lives as long as the sink, so the tracer logger must not add logs to it. - Effect.updateContext(Context.omit(Tracer.ParentSpan)), - ); + const flush = flushSemaphore + .withPermits(1)( + Effect.gen(function* () { + const droppedAtStart = droppedCount; + yield* flushBuffered; + const stats = pendingFlushStats; + pendingFlushStats = { + logicalWriteBytes: 0, + count: 0, + durationMs: 0, + }; + if (stats.count > 0 && options.onFlush) { + yield* options.onFlush(stats).pipe(Effect.ignore); + } + if (droppedCount > 0 && !failureReported) { + failureReported = true; + yield* Effect.logWarning( + "Trace writer could not keep up, dropping records until it recovers", + { + filePath: options.filePath, + }, + ); + } else if (failureReported && !writeFailing && droppedCount === droppedAtStart) { + yield* Effect.logInfo("Trace writes recovered", { + filePath: options.filePath, + droppedCount, + }); + failureReported = false; + droppedCount = 0; + } + }), + ) + .pipe( + // A cancelled write still completes in the OS. Finish it before another + // flush rotates the file or the scope's finalizer drains the remaining records. + Effect.uninterruptible, + Effect.withTracerEnabled(false), + // The timed fiber inherits the makeTraceSink span. That span has ended but + // lives as long as the sink, so the tracer logger must not add logs to it. + Effect.updateContext(Context.omit(Tracer.ParentSpan)), + ); yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore)); yield* Effect.forkScoped( - Effect.sleep(`${options.batchWindowMs} millis`).pipe(Effect.andThen(flush), Effect.forever), + // Flush when push fills a batch or the batch window passes. + Queue.take(flushRequests).pipe( + Effect.timeoutOption(`${options.batchWindowMs} millis`), + Effect.andThen(flush), + Effect.forever, + ), ); return { filePath: options.filePath, push(record) { try { - buffer.push(`${JSON.stringify(record)}\n`); + const text = `${JSON.stringify(record)}\n`; + const bytes = Buffer.byteLength(text); + if (bytes > options.maxBytes) return; + if (bufferBytes + bytes > maxBufferBytes) { + droppedCount++; + return; + } + buffer.push({ text, bytes }); + bufferBytes += bytes; if (buffer.length >= FLUSH_BUFFER_THRESHOLD) { - flushUnsafe(); + Queue.offerUnsafe(flushRequests, undefined); } } catch { return; -- 2.54.0