diff --git a/apps/server/src/observability/EventLoopMonitor.test.ts b/apps/server/src/observability/EventLoopMonitor.test.ts index 1b9fd72049b988a90789b57b6996bf2bb485d262..da990f2b62e16de313c8af81e7c4f8ec71ec8565 100644 --- a/apps/server/src/observability/EventLoopMonitor.test.ts +++ b/apps/server/src/observability/EventLoopMonitor.test.ts @@ -1,6 +1,9 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import * as Tracer from "effect/Tracer"; import * as TestClock from "effect/testing/TestClock"; @@ -25,6 +28,18 @@ const stalled: EventLoopMonitor.EventLoopReadings = { // Over the threshold as read, but not once the resolution is subtracted. const quiet: EventLoopMonitor.EventLoopReadings = { ...stalled, delayMaxNs: ms(2_950) }; +// The first sample covers startup, so the monitor discards it. +const monitor = ( + samples: Array, + stallLog: EventLoopMonitor.StallLogOptions = { filePath: "\0", maxBytes: 1 }, +) => + Layer.build( + EventLoopMonitor.layerWith( + Effect.succeed(Effect.sync(() => samples.shift() ?? quiet)), + stallLog, + ), + ); + describe("EventLoopMonitor", () => { it.effect("records a warning span only for samples that saw a stall", () => Effect.gen(function* () { @@ -36,13 +51,9 @@ describe("EventLoopMonitor", () => { return span; }, }); - // The first sample covers startup, so the monitor discards it. - const samples = [stalled, quiet, stalled]; - yield* Effect.gen(function* () { - yield* Layer.build( - EventLoopMonitor.layerWith(Effect.succeed(Effect.sync(() => samples.shift() ?? quiet))), - ); + // The default stall log path cannot open, and the span must still be recorded. + yield* monitor([stalled, quiet, stalled]); yield* TestClock.adjust("60 seconds"); assert.lengthOf(spans, 0); yield* TestClock.adjust("30 seconds"); @@ -70,6 +81,64 @@ describe("EventLoopMonitor", () => { }), ); + it.effect("keeps each stall as one line in a capped stall log", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-event-loop-" }); + const filePath = `${directory}/event-loop-stalls.ndjson`; + // Room for one line, so each new stall rotates the last one into the backup. + yield* monitor([stalled, stalled, stalled, { ...stalled, delayMaxNs: ms(7_000) }], { + filePath, + maxBytes: 300, + }); + yield* TestClock.adjust("120 seconds"); + + const lines = (yield* fs.readFileString(filePath)).trim().split("\n"); + assert.deepStrictEqual( + lines.map((line) => JSON.parse(line)), + [ + { + time: "1970-01-01T00:02:00.000Z", + delayMaxMs: 6_000, + utilization: 0.18, + cpuUserMs: 310, + cpuSystemMs: 96, + majorPageFaults: 8_412, + minorPageFaults: 20_031, + involuntaryContextSwitches: 57, + rssMb: 1536, + }, + ], + ); + assert.lengthOf((yield* fs.readFileString(`${filePath}.1`)).trim().split("\n"), 1); + assert.isFalse(yield* fs.exists(`${filePath}.2`)); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("warns when a stall cannot be written to the stall log", () => { + const warnings: Array = []; + const logger = Logger.make(({ message }) => { + warnings.push(Array.isArray(message) ? message[0] : message); + }); + return Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + // A directory at the log path opens, but every append to it fails. The cap + // is large so the sink never rotates the directory out of the way. + const filePath = yield* fs.makeTempDirectoryScoped({ prefix: "t3-event-loop-" }); + yield* monitor([stalled, stalled], { filePath, maxBytes: 1024 * 1024 }); + yield* TestClock.adjust("60 seconds"); + assert.deepStrictEqual(warnings, [ + "event loop stalled for 4950 ms", + "event loop stall log write failed", + ]); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.merge(NodeServices.layer, Logger.layer([logger], { mergeWithExisting: false })), + ), + ); + }); + it("ignores delay the loop spent idle, such as a system sleep", () => { // Waking from sleep reads as a long gap, but the loop was idle in poll for it. const asleep: EventLoopMonitor.EventLoopReadings = { diff --git a/apps/server/src/observability/EventLoopMonitor.ts b/apps/server/src/observability/EventLoopMonitor.ts index 13b7b48c4cf7903a8f9eeca4991a1112c0d936d6..544b5ab042d1584b95b0855366b8c5cdc0e6d1ae 100644 --- a/apps/server/src/observability/EventLoopMonitor.ts +++ b/apps/server/src/observability/EventLoopMonitor.ts @@ -1,10 +1,16 @@ // @effect-diagnostics nodeBuiltinImport:off - only node:perf_hooks exposes the event loop delay histogram. import * as NodePerfHooks from "node:perf_hooks"; +import { RotatingFileSink } from "@t3tools/shared/logging"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; +import * as ServerConfig from "../config.ts"; + // Node's delay histogram wakes a native timer every RESOLUTION_MS and records the // gap between wakeups, so an idle loop reads about RESOLUTION_MS and a stall of S // reads between S and S + RESOLUTION_MS. We subtract the resolution, so a delay can @@ -13,6 +19,9 @@ import type * as Scope from "effect/Scope"; const RESOLUTION_MS = 1000; const STALL_THRESHOLD_MS = 2000; const SAMPLE_INTERVAL = "30 seconds"; +// A stall line is about 250 bytes, so the file and its one backup keep the last +// 1000 to 2000 stalls. +const STALL_LOG_MAX_BYTES = 256 * 1024; /** One sample interval as Node reports it. Delay in ns, active time in ms, CPU in µs. */ export interface EventLoopReadings { @@ -83,41 +92,75 @@ export const stallMs = ({ delayMaxNs, activeMs }: EventLoopReadings) => { return delayMs; }; +const encodeStallLine = Schema.encodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)); + +export interface StallLogOptions { + readonly filePath: string; + readonly maxBytes: number; +} + /** * Samples event loop health every 30 s and records a `server.eventLoop.stall` span * with a warning when the loop stalled for more than 2 s, so stalls land in - * the local trace file and Settings > Diagnostics without OTLP. Takes the sampler - * so tests can inject readings. + * the local trace file and Settings > Diagnostics without OTLP. Each stall is + * also appended as one NDJSON line to a small stall log, which keeps stall + * history long after the trace file has rotated. Takes the sampler and log + * options so tests can inject readings and a small cap. */ export const layerWith = ( makeSampler: Effect.Effect, never, Scope.Scope>, + stallLog: StallLogOptions, ) => Layer.effectDiscard( Effect.gen(function* () { const sample = yield* makeSampler; + // A log that cannot open must not stop the monitor; the span still records stalls. + const stallSink = yield* Effect.try( + () => + new RotatingFileSink({ + filePath: stallLog.filePath, + maxBytes: stallLog.maxBytes, + maxFiles: 1, + throwOnError: true, + }), + ).pipe( + Effect.catch(() => + Effect.logWarning("event loop stall log unavailable", { + filePath: stallLog.filePath, + }).pipe(Effect.as(undefined)), + ), + ); const tick = Effect.gen(function* () { const readings = yield* sample; const delayMaxMs = stallMs(readings); if (delayMaxMs === undefined) return; const { utilization, usage, rssBytes } = readings; + const attributes = { + delayMaxMs, + utilization: Math.round(utilization * 100) / 100, + cpuUserMs: Math.round(usage.userCPUTime / 1000), + cpuSystemMs: Math.round(usage.systemCPUTime / 1000), + majorPageFaults: usage.majorPageFault, + minorPageFaults: usage.minorPageFault, + involuntaryContextSwitches: usage.involuntaryContextSwitches, + rssMb: Math.round(rssBytes / 1024 / 1024), + }; // Root, as the stall has no caller to attach to. Warn level keeps it when // T3CODE_TRACE_MIN_LEVEL is raised to cut trace noise. yield* Effect.logWarning(`event loop stalled for ${delayMaxMs} ms`).pipe( - Effect.withSpan("server.eventLoop.stall", { - root: true, - level: "Warn", - attributes: { - delayMaxMs, - utilization: Math.round(utilization * 100) / 100, - cpuUserMs: Math.round(usage.userCPUTime / 1000), - cpuSystemMs: Math.round(usage.systemCPUTime / 1000), - majorPageFaults: usage.majorPageFault, - minorPageFaults: usage.minorPageFault, - involuntaryContextSwitches: usage.involuntaryContextSwitches, - rssMb: Math.round(rssBytes / 1024 / 1024), - }, - }), + Effect.withSpan("server.eventLoop.stall", { root: true, level: "Warn", attributes }), ); + if (stallSink !== undefined) { + const time = DateTime.formatIso(yield* DateTime.now); + yield* encodeStallLine({ time, ...attributes }).pipe( + Effect.flatMap((line) => Effect.try(() => stallSink.write(`${line}\n`))), + Effect.catch(() => + Effect.logWarning("event loop stall log write failed", { + filePath: stallLog.filePath, + }), + ), + ); + } }); const wait = Effect.sleep(SAMPLE_INTERVAL); // The layer builds before the rest of the server, so the first sample covers @@ -132,4 +175,14 @@ export const layerWith = ( }), ); -export const layer = layerWith(makeNodeSampler); +/** Writes the stall log next to the server trace file, as `event-loop-stalls.ndjson`. */ +export const layer = Layer.unwrap( + Effect.gen(function* () { + const { serverTracePath } = yield* ServerConfig.ServerConfig; + const path = yield* Path.Path; + return layerWith(makeNodeSampler, { + filePath: path.join(path.dirname(serverTracePath), "event-loop-stalls.ndjson"), + maxBytes: STALL_LOG_MAX_BYTES, + }); + }), +); diff --git a/apps/server/src/observability/Observability.ts b/apps/server/src/observability/Observability.ts index ab078fb121aa64af34fd490ee2ea136da18eb251..9ae9875749b243c62c2dbc2368b64a717b63ce7a 100644 --- a/apps/server/src/observability/Observability.ts +++ b/apps/server/src/observability/Observability.ts @@ -9,6 +9,7 @@ import * as Tracer from "effect/Tracer"; import * as OtlpExporter from "effect/observability/OtlpExporter"; import * as OtlpMetrics from "effect/observability/OtlpMetrics"; import * as OtlpTracer from "effect/observability/OtlpTracer"; +import * as Statement from "effect/sql/Statement"; import * as ServerConfig from "../config.ts"; import * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts"; @@ -30,6 +31,9 @@ export const layer = Layer.unwrap( const layerTraceReferences = Layer.mergeAll( Layer.succeed(Tracer.MinimumTraceLevel, config.traceMinLevel), Layer.succeed(References.TracerTimingEnabled, config.traceTimingEnabled), + // Keeps each SQL statement's span current while it waits for a connection, + // so the client can record the lock wait on it. + Layer.succeed(Statement.SpanPropagationEnabled, true), HttpObservability.layer, ); diff --git a/docs/operations/observability.md b/docs/operations/observability.md index b434eb6e396cfc3092f38118e5242c7114df9b02..eedb8732ea2717b9214a822c75896af8a2841935 100644 --- a/docs/operations/observability.md +++ b/docs/operations/observability.md @@ -56,6 +56,13 @@ records instead carry OTLP resource, scope, and optional status fields. The `TraceRecord`, `EffectTraceRecord`, and `OtlpTraceRecord` schemas live in `packages/shared/src/observability.ts`. +SQL statement spans (named `sqlite`) and `sql.transaction` spans carry `db.connection` and +`db.lock_wait_ms`. `db.connection` is `writer` for the main connection, which runs on the event loop, +or `reader` for the read-only connection in a worker thread. `db.lock_wait_ms` is the part of the +span spent waiting for that connection, behind other SQL. That wait does not block the event loop, +so subtract it before you blame a slow span on the query. Statements inside a transaction use its +connection and do not carry these attributes. + DPoP proof failures include the safe `environment.dpop.failure_code` span attribute. A `time_window` failure means that a signed proof was too old or too far in the future for the environment server's allowed window. It can point to @@ -95,6 +102,11 @@ the loop stalled for more than 2 s since the previous sample, it records a root `T3CODE_TRACE_MIN_LEVEL` is `Warn`. The warning shows in Settings > Diagnostics unless OTLP logs are on. The span time is when the sample ran, not when the stall happened. +The trace file rotates in well under an hour on a busy server, so each stall is also appended as one +NDJSON line to `event-loop-stalls.ndjson`, next to the trace file. A line has `time` and the span +attributes. The file keeps the last 1000 or more stalls, with one rotated backup +(`event-loop-stalls.ndjson.1`). Use it to see how often a server stalls over days. + Some delay is not recorded: - `delayMaxMs` is the longest stall, and can undercount it by up to 1 s. The 2 s threshold applies to diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index c5001907e9e96b875137ee54859b1c668a791364..98dbafeb5d0a7a9e8f57ca2de4de4fc05dc5c4f2 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -10,7 +10,9 @@ import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import * as TestClock from "effect/testing/TestClock"; +import * as Tracer from "effect/Tracer"; import { vi } from "vite-plus/test"; import * as SqliteClient from "./nodeSqliteClient.ts"; @@ -411,6 +413,99 @@ it.effect("keeps read-only work inside a write transaction on the writer", () => ), ); +it.effect("labels SQL spans with their connection and how long they waited for its lock", () => { + const spans: Array = []; + const tracer = Tracer.make({ + span: (options) => { + const span = new Tracer.NativeSpan(options); + spans.push(span); + return span; + }, + }); + return withReaderClient((sql) => + Effect.gen(function* () { + const held = yield* Deferred.make(); + const release = yield* Deferred.make(); + const holder = yield* Effect.forkChild( + sql.withTransaction( + Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))), + ), + ); + yield* Deferred.await(held); + // Each waiter runs until it blocks on the writer lock, before the clock moves. + const statement = yield* Effect.forkChild(sql`SELECT 1`, { startImmediately: true }); + const transaction = yield* Effect.forkChild(sql.withTransaction(countEntries(sql)), { + startImmediately: true, + }); + yield* TestClock.adjust("25 millis"); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(holder); + yield* Fiber.join(statement); + yield* Fiber.join(transaction); + yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + }), + ).pipe( + // The server turns this on, so a statement's span is current while it waits. + Effect.provideService(Statement.SpanPropagationEnabled, true), + Effect.withTracer(tracer), + Effect.map(() => { + // Statements inside a transaction reuse its connection, so only the + // statements and transactions that took a lock are labeled. + const labeled = spans + .filter((span) => span.attributes.has("db.connection")) + .map((span) => [ + span.name, + span.attributes.get("db.connection"), + span.attributes.get("db.lock_wait_ms"), + ]); + assert.deepStrictEqual(labeled.slice(3), [ + ["sql.transaction", "writer", 0], + ["sqlite", "writer", 25], + ["sql.transaction", "writer", 25], + ["sql.transaction", "reader", 0], + ]); + }), + ); +}); + +it.effect("labels only the span of the statement or transaction that took the lock", () => { + const spans: Array = []; + const tracer = Tracer.make({ + span: (options) => { + const span = new Tracer.NativeSpan(options); + spans.push(span); + return span; + }, + }); + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const transaction = sql.withTransaction(sql`SELECT 1`); + yield* transaction; + // Run again nested, the same Effect reuses the outer connection and takes no lock. + yield* sql.withTransaction(transaction); + // With tracing off, Statement does not make its span current. + yield* sql`SELECT 2`.pipe(Effect.withTracerEnabled(false)); + }).pipe( + Effect.withSpan("caller"), + Effect.provide(SqliteClient.layer({ filename: ":memory:" })), + Effect.provideService(Statement.SpanPropagationEnabled, true), + Effect.withTracer(tracer), + Effect.map(() => { + assert.deepStrictEqual( + spans + .filter((span) => span.name === "caller" || span.name === "sql.transaction") + .map((span) => [span.name, span.attributes.get("db.connection")]), + [ + ["caller", undefined], + ["sql.transaction", "writer"], + ["sql.transaction", "writer"], + ["sql.transaction", undefined], + ], + ); + }), + ); +}); + it.effect("keeps large parameters off the reader worker's cached statements", () => withReaderClient((sql) => Effect.gen(function* () { diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 996b8618bdd4589702c8ef6222ae2a29640cb416..f0a93b1c21fddce23eb1c867966419f1f9706d36 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -9,6 +9,7 @@ import * as NodeUtil from "node:util"; import * as NodeWorkerThreads from "node:worker_threads"; import * as Cache from "effect/Cache"; +import * as Clock from "effect/Clock"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -17,11 +18,13 @@ import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; +import * as References from "effect/References"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Context from "effect/Context"; import * as Stream from "effect/Stream"; +import * as Tracer from "effect/Tracer"; import * as Reactivity from "effect/reactivity/Reactivity"; import * as Client from "effect/sql/SqlClient"; import type { Connection } from "effect/sql/SqlConnection"; @@ -72,6 +75,33 @@ const ReadOnly = Context.Reference("@t3tools/shared/nodeSqliteClient/Re export const readOnly = (effect: Effect.Effect): Effect.Effect => Effect.provideService(effect, ReadOnly, true); +/** + * Which connection ran the SQL, and how long it waited for that connection's + * lock. The wait is time queued behind other SQL, not time the event loop was + * blocked. + */ +interface Acquired { + readonly connection: "writer" | "reader"; + readonly lockWaitMs: number; +} + +const annotateAcquired = (fiber: Fiber.Fiber, acquired: Acquired | undefined) => { + const span = Context.getOrUndefined(fiber.context, Tracer.ParentSpan); + if (span?._tag !== "Span" || acquired === undefined) return; + span.attribute("db.connection", acquired.connection); + span.attribute("db.lock_wait_ms", acquired.lockWaitMs); +}; + +/** + * Provided by the client's `withTransaction`. Effect acquires a transaction's + * connection before it makes the `sql.transaction` span current, so the acquire + * leaves its result here and the transaction body puts it on the span. + */ +const TransactionAcquire = Context.Reference<{ acquired?: Acquired } | undefined>( + "@t3tools/shared/nodeSqliteClient/TransactionAcquire", + { defaultValue: () => undefined }, +); + export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError()( "UnsupportedNodeSqliteVersionError", { @@ -785,7 +815,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( options.readerWorker && isFileDatabase ? yield* makeReader(options) : undefined; const readerSemaphore = yield* Semaphore.make(1); - const writer = { lock: semaphore, connection }; + const writer = { name: "writer" as const, lock: semaphore, connection }; // Picks the reader only for `readOnly` work, and the writer whenever the // reader is turned off. const route = Effect.withFiber((fiber) => @@ -793,28 +823,59 @@ const make = Effect.fn("makeWithDatabase")(function* ( ? Effect.map(acquireReader, (readerConnection) => readerConnection === undefined ? writer - : { lock: readerSemaphore, connection: readerConnection }, + : { name: "reader" as const, lock: readerSemaphore, connection: readerConnection }, ) : Effect.succeed(writer), ); + const acquired = ( + target: { readonly name: Acquired["connection"] }, + clock: Clock.Clock, + start: bigint, + ): Acquired => ({ + connection: target.name, + lockWaitMs: Number((clock.monotonicTimeNanosUnsafe() - start) / 1000n) / 1000, + }); + const acquirer = Effect.flatMap(route, (target) => - target.lock.withPermits(1)(Effect.succeed(target.connection)), + Effect.withFiber((fiber) => { + // The statement's span is current here only when it propagates, which + // Statement skips while tracing is off. Otherwise the current span is + // the caller's, so leave it alone. + if ( + !fiber.getRef(References.TracerEnabled) || + !fiber.getRef(Statement.SpanPropagationEnabled) + ) { + return target.lock.withPermits(1)(Effect.succeed(target.connection)); + } + const clock = fiber.getRef(Clock.Clock); + const start = clock.monotonicTimeNanosUnsafe(); + return target.lock.withPermits(1)( + Effect.sync(() => { + annotateAcquired(fiber, acquired(target, clock, start)); + return target.connection; + }), + ); + }), ); const transactionAcquirer = Effect.flatMap(route, (target) => Effect.uninterruptibleMask((restore) => { const fiber = Fiber.getCurrent()!; const scope = Context.getUnsafe(fiber.context, Scope.Scope); + const clock = fiber.getRef(Clock.Clock); + const start = clock.monotonicTimeNanosUnsafe(); return Effect.as( - Effect.tap(restore(target.lock.take(1)), () => - Scope.addFinalizer(scope, target.lock.release(1)), - ), + Effect.tap(restore(target.lock.take(1)), () => { + const pending = fiber.getRef(TransactionAcquire); + if (pending !== undefined) pending.acquired = acquired(target, clock, start); + return Scope.addFinalizer(scope, target.lock.release(1)); + }), target.connection, ); }), ); - return yield* Client.make({ + const client = yield* Client.make({ acquirer, compiler, transactionAcquirer, @@ -831,6 +892,22 @@ const make = Effect.fn("makeWithDatabase")(function* ( ], transformRows, }); + const withTransaction = client.withTransaction; + return Object.assign(client, { + // Each run gets its own holder, so a reused transaction Effect never + // carries an earlier run's attributes. + withTransaction: (effect: Effect.Effect) => + Effect.suspend(() => { + const pending: { acquired?: Acquired } = {}; + // The body runs with the `sql.transaction` span current. A nested + // transaction acquires nothing, so its span gets no attributes. + const body = Effect.withFiber((fiber) => { + annotateAcquired(fiber, pending.acquired); + return effect; + }); + return withTransaction(body).pipe(Effect.provideService(TransactionAcquire, pending)); + }), + }); }); export const layer = (config: SqliteClientConfig): Layer.Layer =>