feat(observability): SQL spans show lock wait and connection; stalls keep their own history

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<EventLoopMonitor.EventLoopReadings>,
+  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<unknown> = [];
+    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 = {

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<Effect.Effect<EventLoopReadings>, 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,
+    });
+  }),
+);

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,
     );
 

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

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";
@@ -407,6 +409,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<Tracer.NativeSpan> = [];
+  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<void>();
+      const release = yield* Deferred.make<void>();
+      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<Tracer.NativeSpan> = [];
+  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* () {