authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 05:55:36-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 05:55:36-07:00
log28732b4787261609c4d20b82b244cb6eef0c442e
tree86101d5a59b28df9a657ea8cbe6578d5a45d3226
parent36f91c45687f97fc8e12ff89eaa8cff6f23abe8b
signature Signed by PGP key B5690EEEBB952194

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

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

6 files changed, 340 insertions(+), 30 deletions(-)

apps/server/src/observability/EventLoopMonitor.test.ts+75-6
...@@ -1,6 +1,9 @@...@@ -1,6 +1,9 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
1import { assert, describe, it } from "@effect/vitest";2import { assert, describe, it } from "@effect/vitest";
2import * as Effect from "effect/Effect";3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
3import * as Layer from "effect/Layer";5import * as Layer from "effect/Layer";
6import * as Logger from "effect/Logger";
4import * as Tracer from "effect/Tracer";7import * as Tracer from "effect/Tracer";
5import * as TestClock from "effect/testing/TestClock";8import * as TestClock from "effect/testing/TestClock";
69
...@@ -25,6 +28,18 @@ const stalled: EventLoopMonitor.EventLoopReadings = {...@@ -25,6 +28,18 @@ const stalled: EventLoopMonitor.EventLoopReadings = {
25// Over the threshold as read, but not once the resolution is subtracted.28// Over the threshold as read, but not once the resolution is subtracted.
26const quiet: EventLoopMonitor.EventLoopReadings = { ...stalled, delayMaxNs: ms(2_950) };29const quiet: EventLoopMonitor.EventLoopReadings = { ...stalled, delayMaxNs: ms(2_950) };
2730
31// The first sample covers startup, so the monitor discards it.
32const monitor = (
33 samples: Array<EventLoopMonitor.EventLoopReadings>,
34 stallLog: EventLoopMonitor.StallLogOptions = { filePath: "\0", maxBytes: 1 },
35) =>
36 Layer.build(
37 EventLoopMonitor.layerWith(
38 Effect.succeed(Effect.sync(() => samples.shift() ?? quiet)),
39 stallLog,
40 ),
41 );
42
28describe("EventLoopMonitor", () => {43describe("EventLoopMonitor", () => {
29 it.effect("records a warning span only for samples that saw a stall", () =>44 it.effect("records a warning span only for samples that saw a stall", () =>
30 Effect.gen(function* () {45 Effect.gen(function* () {
...@@ -36,13 +51,9 @@ describe("EventLoopMonitor", () => {...@@ -36,13 +51,9 @@ describe("EventLoopMonitor", () => {
36 return span;51 return span;
37 },52 },
38 });53 });
39 // The first sample covers startup, so the monitor discards it.
40 const samples = [stalled, quiet, stalled];
41
42 yield* Effect.gen(function* () {54 yield* Effect.gen(function* () {
43 yield* Layer.build(55 // The default stall log path cannot open, and the span must still be recorded.
44 EventLoopMonitor.layerWith(Effect.succeed(Effect.sync(() => samples.shift() ?? quiet))),56 yield* monitor([stalled, quiet, stalled]);
45 );
46 yield* TestClock.adjust("60 seconds");57 yield* TestClock.adjust("60 seconds");
47 assert.lengthOf(spans, 0);58 assert.lengthOf(spans, 0);
48 yield* TestClock.adjust("30 seconds");59 yield* TestClock.adjust("30 seconds");
...@@ -70,6 +81,64 @@ describe("EventLoopMonitor", () => {...@@ -70,6 +81,64 @@ describe("EventLoopMonitor", () => {
70 }),81 }),
71 );82 );
7283
84 it.effect("keeps each stall as one line in a capped stall log", () =>
85 Effect.gen(function* () {
86 const fs = yield* FileSystem.FileSystem;
87 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-event-loop-" });
88 const filePath = `${directory}/event-loop-stalls.ndjson`;
89 // Room for one line, so each new stall rotates the last one into the backup.
90 yield* monitor([stalled, stalled, stalled, { ...stalled, delayMaxNs: ms(7_000) }], {
91 filePath,
92 maxBytes: 300,
93 });
94 yield* TestClock.adjust("120 seconds");
95
96 const lines = (yield* fs.readFileString(filePath)).trim().split("\n");
97 assert.deepStrictEqual(
98 lines.map((line) => JSON.parse(line)),
99 [
100 {
101 time: "1970-01-01T00:02:00.000Z",
102 delayMaxMs: 6_000,
103 utilization: 0.18,
104 cpuUserMs: 310,
105 cpuSystemMs: 96,
106 majorPageFaults: 8_412,
107 minorPageFaults: 20_031,
108 involuntaryContextSwitches: 57,
109 rssMb: 1536,
110 },
111 ],
112 );
113 assert.lengthOf((yield* fs.readFileString(`${filePath}.1`)).trim().split("\n"), 1);
114 assert.isFalse(yield* fs.exists(`${filePath}.2`));
115 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
116 );
117
118 it.effect("warns when a stall cannot be written to the stall log", () => {
119 const warnings: Array<unknown> = [];
120 const logger = Logger.make(({ message }) => {
121 warnings.push(Array.isArray(message) ? message[0] : message);
122 });
123 return Effect.gen(function* () {
124 const fs = yield* FileSystem.FileSystem;
125 // A directory at the log path opens, but every append to it fails. The cap
126 // is large so the sink never rotates the directory out of the way.
127 const filePath = yield* fs.makeTempDirectoryScoped({ prefix: "t3-event-loop-" });
128 yield* monitor([stalled, stalled], { filePath, maxBytes: 1024 * 1024 });
129 yield* TestClock.adjust("60 seconds");
130 assert.deepStrictEqual(warnings, [
131 "event loop stalled for 4950 ms",
132 "event loop stall log write failed",
133 ]);
134 }).pipe(
135 Effect.scoped,
136 Effect.provide(
137 Layer.merge(NodeServices.layer, Logger.layer([logger], { mergeWithExisting: false })),
138 ),
139 );
140 });
141
73 it("ignores delay the loop spent idle, such as a system sleep", () => {142 it("ignores delay the loop spent idle, such as a system sleep", () => {
74 // Waking from sleep reads as a long gap, but the loop was idle in poll for it.143 // Waking from sleep reads as a long gap, but the loop was idle in poll for it.
75 const asleep: EventLoopMonitor.EventLoopReadings = {144 const asleep: EventLoopMonitor.EventLoopReadings = {
apps/server/src/observability/EventLoopMonitor.ts+70-17
...@@ -1,10 +1,16 @@...@@ -1,10 +1,16 @@
1// @effect-diagnostics nodeBuiltinImport:off - only node:perf_hooks exposes the event loop delay histogram.1// @effect-diagnostics nodeBuiltinImport:off - only node:perf_hooks exposes the event loop delay histogram.
2import * as NodePerfHooks from "node:perf_hooks";2import * as NodePerfHooks from "node:perf_hooks";
33
4import { RotatingFileSink } from "@t3tools/shared/logging";
5import * as DateTime from "effect/DateTime";
4import * as Effect from "effect/Effect";6import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";7import * as Layer from "effect/Layer";
8import * as Path from "effect/Path";
9import * as Schema from "effect/Schema";
6import type * as Scope from "effect/Scope";10import type * as Scope from "effect/Scope";
711
12import * as ServerConfig from "../config.ts";
13
8// Node's delay histogram wakes a native timer every RESOLUTION_MS and records the14// Node's delay histogram wakes a native timer every RESOLUTION_MS and records the
9// gap between wakeups, so an idle loop reads about RESOLUTION_MS and a stall of S15// gap between wakeups, so an idle loop reads about RESOLUTION_MS and a stall of S
10// reads between S and S + RESOLUTION_MS. We subtract the resolution, so a delay can16// 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";...@@ -13,6 +19,9 @@ import type * as Scope from "effect/Scope";
13const RESOLUTION_MS = 1000;19const RESOLUTION_MS = 1000;
14const STALL_THRESHOLD_MS = 2000;20const STALL_THRESHOLD_MS = 2000;
15const SAMPLE_INTERVAL = "30 seconds";21const SAMPLE_INTERVAL = "30 seconds";
22// A stall line is about 250 bytes, so the file and its one backup keep the last
23// 1000 to 2000 stalls.
24const STALL_LOG_MAX_BYTES = 256 * 1024;
1625
17/** One sample interval as Node reports it. Delay in ns, active time in ms, CPU in µs. */26/** One sample interval as Node reports it. Delay in ns, active time in ms, CPU in µs. */
18export interface EventLoopReadings {27export interface EventLoopReadings {
...@@ -83,41 +92,75 @@ export const stallMs = ({ delayMaxNs, activeMs }: EventLoopReadings) => {...@@ -83,41 +92,75 @@ export const stallMs = ({ delayMaxNs, activeMs }: EventLoopReadings) => {
83 return delayMs;92 return delayMs;
84};93};
8594
95const encodeStallLine = Schema.encodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
96
97export interface StallLogOptions {
98 readonly filePath: string;
99 readonly maxBytes: number;
100}
101
86/**102/**
87 * Samples event loop health every 30 s and records a `server.eventLoop.stall` span103 * Samples event loop health every 30 s and records a `server.eventLoop.stall` span
88 * with a warning when the loop stalled for more than 2 s, so stalls land in104 * with a warning when the loop stalled for more than 2 s, so stalls land in
89 * the local trace file and Settings > Diagnostics without OTLP. Takes the sampler105 * the local trace file and Settings > Diagnostics without OTLP. Each stall is
90 * so tests can inject readings.106 * also appended as one NDJSON line to a small stall log, which keeps stall
107 * history long after the trace file has rotated. Takes the sampler and log
108 * options so tests can inject readings and a small cap.
91 */109 */
92export const layerWith = (110export const layerWith = (
93 makeSampler: Effect.Effect<Effect.Effect<EventLoopReadings>, never, Scope.Scope>,111 makeSampler: Effect.Effect<Effect.Effect<EventLoopReadings>, never, Scope.Scope>,
112 stallLog: StallLogOptions,
94) =>113) =>
95 Layer.effectDiscard(114 Layer.effectDiscard(
96 Effect.gen(function* () {115 Effect.gen(function* () {
97 const sample = yield* makeSampler;116 const sample = yield* makeSampler;
117 // A log that cannot open must not stop the monitor; the span still records stalls.
118 const stallSink = yield* Effect.try(
119 () =>
120 new RotatingFileSink({
121 filePath: stallLog.filePath,
122 maxBytes: stallLog.maxBytes,
123 maxFiles: 1,
124 throwOnError: true,
125 }),
126 ).pipe(
127 Effect.catch(() =>
128 Effect.logWarning("event loop stall log unavailable", {
129 filePath: stallLog.filePath,
130 }).pipe(Effect.as(undefined)),
131 ),
132 );
98 const tick = Effect.gen(function* () {133 const tick = Effect.gen(function* () {
99 const readings = yield* sample;134 const readings = yield* sample;
100 const delayMaxMs = stallMs(readings);135 const delayMaxMs = stallMs(readings);
101 if (delayMaxMs === undefined) return;136 if (delayMaxMs === undefined) return;
102 const { utilization, usage, rssBytes } = readings;137 const { utilization, usage, rssBytes } = readings;
138 const attributes = {
139 delayMaxMs,
140 utilization: Math.round(utilization * 100) / 100,
141 cpuUserMs: Math.round(usage.userCPUTime / 1000),
142 cpuSystemMs: Math.round(usage.systemCPUTime / 1000),
143 majorPageFaults: usage.majorPageFault,
144 minorPageFaults: usage.minorPageFault,
145 involuntaryContextSwitches: usage.involuntaryContextSwitches,
146 rssMb: Math.round(rssBytes / 1024 / 1024),
147 };
103 // Root, as the stall has no caller to attach to. Warn level keeps it when148 // Root, as the stall has no caller to attach to. Warn level keeps it when
104 // T3CODE_TRACE_MIN_LEVEL is raised to cut trace noise.149 // T3CODE_TRACE_MIN_LEVEL is raised to cut trace noise.
105 yield* Effect.logWarning(`event loop stalled for ${delayMaxMs} ms`).pipe(150 yield* Effect.logWarning(`event loop stalled for ${delayMaxMs} ms`).pipe(
106 Effect.withSpan("server.eventLoop.stall", {151 Effect.withSpan("server.eventLoop.stall", { root: true, level: "Warn", attributes }),
107 root: true,
108 level: "Warn",
109 attributes: {
110 delayMaxMs,
111 utilization: Math.round(utilization * 100) / 100,
112 cpuUserMs: Math.round(usage.userCPUTime / 1000),
113 cpuSystemMs: Math.round(usage.systemCPUTime / 1000),
114 majorPageFaults: usage.majorPageFault,
115 minorPageFaults: usage.minorPageFault,
116 involuntaryContextSwitches: usage.involuntaryContextSwitches,
117 rssMb: Math.round(rssBytes / 1024 / 1024),
118 },
119 }),
120 );152 );
153 if (stallSink !== undefined) {
154 const time = DateTime.formatIso(yield* DateTime.now);
155 yield* encodeStallLine({ time, ...attributes }).pipe(
156 Effect.flatMap((line) => Effect.try(() => stallSink.write(`${line}\n`))),
157 Effect.catch(() =>
158 Effect.logWarning("event loop stall log write failed", {
159 filePath: stallLog.filePath,
160 }),
161 ),
162 );
163 }
121 });164 });
122 const wait = Effect.sleep(SAMPLE_INTERVAL);165 const wait = Effect.sleep(SAMPLE_INTERVAL);
123 // The layer builds before the rest of the server, so the first sample covers166 // The layer builds before the rest of the server, so the first sample covers
...@@ -132,4 +175,14 @@ export const layerWith = (...@@ -132,4 +175,14 @@ export const layerWith = (
132 }),175 }),
133 );176 );
134177
135export const layer = layerWith(makeNodeSampler);178/** Writes the stall log next to the server trace file, as `event-loop-stalls.ndjson`. */
179export const layer = Layer.unwrap(
180 Effect.gen(function* () {
181 const { serverTracePath } = yield* ServerConfig.ServerConfig;
182 const path = yield* Path.Path;
183 return layerWith(makeNodeSampler, {
184 filePath: path.join(path.dirname(serverTracePath), "event-loop-stalls.ndjson"),
185 maxBytes: STALL_LOG_MAX_BYTES,
186 });
187 }),
188);
apps/server/src/observability/Observability.ts+4
...@@ -9,6 +9,7 @@ import * as Tracer from "effect/Tracer";...@@ -9,6 +9,7 @@ import * as Tracer from "effect/Tracer";
9import * as OtlpExporter from "effect/observability/OtlpExporter";9import * as OtlpExporter from "effect/observability/OtlpExporter";
10import * as OtlpMetrics from "effect/observability/OtlpMetrics";10import * as OtlpMetrics from "effect/observability/OtlpMetrics";
11import * as OtlpTracer from "effect/observability/OtlpTracer";11import * as OtlpTracer from "effect/observability/OtlpTracer";
12import * as Statement from "effect/sql/Statement";
1213
13import * as ServerConfig from "../config.ts";14import * as ServerConfig from "../config.ts";
14import * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts";15import * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts";
...@@ -30,6 +31,9 @@ export const layer = Layer.unwrap(...@@ -30,6 +31,9 @@ export const layer = Layer.unwrap(
30 const layerTraceReferences = Layer.mergeAll(31 const layerTraceReferences = Layer.mergeAll(
31 Layer.succeed(Tracer.MinimumTraceLevel, config.traceMinLevel),32 Layer.succeed(Tracer.MinimumTraceLevel, config.traceMinLevel),
32 Layer.succeed(References.TracerTimingEnabled, config.traceTimingEnabled),33 Layer.succeed(References.TracerTimingEnabled, config.traceTimingEnabled),
34 // Keeps each SQL statement's span current while it waits for a connection,
35 // so the client can record the lock wait on it.
36 Layer.succeed(Statement.SpanPropagationEnabled, true),
33 HttpObservability.layer,37 HttpObservability.layer,
34 );38 );
3539
docs/operations/observability.md+12
...@@ -56,6 +56,13 @@ records instead carry OTLP resource, scope, and optional status fields....@@ -56,6 +56,13 @@ records instead carry OTLP resource, scope, and optional status fields.
56The `TraceRecord`, `EffectTraceRecord`, and `OtlpTraceRecord` schemas live in56The `TraceRecord`, `EffectTraceRecord`, and `OtlpTraceRecord` schemas live in
57`packages/shared/src/observability.ts`.57`packages/shared/src/observability.ts`.
5858
59SQL statement spans (named `sqlite`) and `sql.transaction` spans carry `db.connection` and
60`db.lock_wait_ms`. `db.connection` is `writer` for the main connection, which runs on the event loop,
61or `reader` for the read-only connection in a worker thread. `db.lock_wait_ms` is the part of the
62span spent waiting for that connection, behind other SQL. That wait does not block the event loop,
63so subtract it before you blame a slow span on the query. Statements inside a transaction use its
64connection and do not carry these attributes.
65
59DPoP proof failures include the safe `environment.dpop.failure_code` span66DPoP proof failures include the safe `environment.dpop.failure_code` span
60attribute. A `time_window` failure means that a signed proof was too old or too67attribute. A `time_window` failure means that a signed proof was too old or too
61far in the future for the environment server's allowed window. It can point to68far 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...@@ -95,6 +102,11 @@ the loop stalled for more than 2 s since the previous sample, it records a root
95`T3CODE_TRACE_MIN_LEVEL` is `Warn`. The warning shows in Settings > Diagnostics unless OTLP logs are102`T3CODE_TRACE_MIN_LEVEL` is `Warn`. The warning shows in Settings > Diagnostics unless OTLP logs are
96on. The span time is when the sample ran, not when the stall happened.103on. The span time is when the sample ran, not when the stall happened.
97104
105The trace file rotates in well under an hour on a busy server, so each stall is also appended as one
106NDJSON line to `event-loop-stalls.ndjson`, next to the trace file. A line has `time` and the span
107attributes. The file keeps the last 1000 or more stalls, with one rotated backup
108(`event-loop-stalls.ndjson.1`). Use it to see how often a server stalls over days.
109
98Some delay is not recorded:110Some delay is not recorded:
99111
100- `delayMaxMs` is the longest stall, and can undercount it by up to 1 s. The 2 s threshold applies to112- `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+95
...@@ -10,7 +10,9 @@ import * as Layer from "effect/Layer";...@@ -10,7 +10,9 @@ import * as Layer from "effect/Layer";
10import * as Logger from "effect/Logger";10import * as Logger from "effect/Logger";
11import * as Path from "effect/Path";11import * as Path from "effect/Path";
12import * as SqlClient from "effect/sql/SqlClient";12import * as SqlClient from "effect/sql/SqlClient";
13import * as Statement from "effect/sql/Statement";
13import * as TestClock from "effect/testing/TestClock";14import * as TestClock from "effect/testing/TestClock";
15import * as Tracer from "effect/Tracer";
14import { vi } from "vite-plus/test";16import { vi } from "vite-plus/test";
1517
16import * as SqliteClient from "./nodeSqliteClient.ts";18import * as SqliteClient from "./nodeSqliteClient.ts";
...@@ -411,6 +413,99 @@ it.effect("keeps read-only work inside a write transaction on the writer", () =>...@@ -411,6 +413,99 @@ it.effect("keeps read-only work inside a write transaction on the writer", () =>
411 ),413 ),
412);414);
413415
416it.effect("labels SQL spans with their connection and how long they waited for its lock", () => {
417 const spans: Array<Tracer.NativeSpan> = [];
418 const tracer = Tracer.make({
419 span: (options) => {
420 const span = new Tracer.NativeSpan(options);
421 spans.push(span);
422 return span;
423 },
424 });
425 return withReaderClient((sql) =>
426 Effect.gen(function* () {
427 const held = yield* Deferred.make<void>();
428 const release = yield* Deferred.make<void>();
429 const holder = yield* Effect.forkChild(
430 sql.withTransaction(
431 Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))),
432 ),
433 );
434 yield* Deferred.await(held);
435 // Each waiter runs until it blocks on the writer lock, before the clock moves.
436 const statement = yield* Effect.forkChild(sql`SELECT 1`, { startImmediately: true });
437 const transaction = yield* Effect.forkChild(sql.withTransaction(countEntries(sql)), {
438 startImmediately: true,
439 });
440 yield* TestClock.adjust("25 millis");
441 yield* Deferred.succeed(release, undefined);
442 yield* Fiber.join(holder);
443 yield* Fiber.join(statement);
444 yield* Fiber.join(transaction);
445 yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly);
446 }),
447 ).pipe(
448 // The server turns this on, so a statement's span is current while it waits.
449 Effect.provideService(Statement.SpanPropagationEnabled, true),
450 Effect.withTracer(tracer),
451 Effect.map(() => {
452 // Statements inside a transaction reuse its connection, so only the
453 // statements and transactions that took a lock are labeled.
454 const labeled = spans
455 .filter((span) => span.attributes.has("db.connection"))
456 .map((span) => [
457 span.name,
458 span.attributes.get("db.connection"),
459 span.attributes.get("db.lock_wait_ms"),
460 ]);
461 assert.deepStrictEqual(labeled.slice(3), [
462 ["sql.transaction", "writer", 0],
463 ["sqlite", "writer", 25],
464 ["sql.transaction", "writer", 25],
465 ["sql.transaction", "reader", 0],
466 ]);
467 }),
468 );
469});
470
471it.effect("labels only the span of the statement or transaction that took the lock", () => {
472 const spans: Array<Tracer.NativeSpan> = [];
473 const tracer = Tracer.make({
474 span: (options) => {
475 const span = new Tracer.NativeSpan(options);
476 spans.push(span);
477 return span;
478 },
479 });
480 return Effect.gen(function* () {
481 const sql = yield* SqlClient.SqlClient;
482 const transaction = sql.withTransaction(sql`SELECT 1`);
483 yield* transaction;
484 // Run again nested, the same Effect reuses the outer connection and takes no lock.
485 yield* sql.withTransaction(transaction);
486 // With tracing off, Statement does not make its span current.
487 yield* sql`SELECT 2`.pipe(Effect.withTracerEnabled(false));
488 }).pipe(
489 Effect.withSpan("caller"),
490 Effect.provide(SqliteClient.layer({ filename: ":memory:" })),
491 Effect.provideService(Statement.SpanPropagationEnabled, true),
492 Effect.withTracer(tracer),
493 Effect.map(() => {
494 assert.deepStrictEqual(
495 spans
496 .filter((span) => span.name === "caller" || span.name === "sql.transaction")
497 .map((span) => [span.name, span.attributes.get("db.connection")]),
498 [
499 ["caller", undefined],
500 ["sql.transaction", "writer"],
501 ["sql.transaction", "writer"],
502 ["sql.transaction", undefined],
503 ],
504 );
505 }),
506 );
507});
508
414it.effect("keeps large parameters off the reader worker's cached statements", () =>509it.effect("keeps large parameters off the reader worker's cached statements", () =>
415 withReaderClient((sql) =>510 withReaderClient((sql) =>
416 Effect.gen(function* () {511 Effect.gen(function* () {
packages/shared/src/nodeSqliteClient.ts+84-7
...@@ -9,6 +9,7 @@ import * as NodeUtil from "node:util";...@@ -9,6 +9,7 @@ import * as NodeUtil from "node:util";
9import * as NodeWorkerThreads from "node:worker_threads";9import * as NodeWorkerThreads from "node:worker_threads";
1010
11import * as Cache from "effect/Cache";11import * as Cache from "effect/Cache";
12import * as Clock from "effect/Clock";
12import * as Duration from "effect/Duration";13import * as Duration from "effect/Duration";
13import * as Effect from "effect/Effect";14import * as Effect from "effect/Effect";
14import * as Exit from "effect/Exit";15import * as Exit from "effect/Exit";
...@@ -17,11 +18,13 @@ import { identity } from "effect/Function";...@@ -17,11 +18,13 @@ import { identity } from "effect/Function";
17import * as Layer from "effect/Layer";18import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";19import * as Option from "effect/Option";
19import * as Predicate from "effect/Predicate";20import * as Predicate from "effect/Predicate";
21import * as References from "effect/References";
20import * as Schema from "effect/Schema";22import * as Schema from "effect/Schema";
21import * as Scope from "effect/Scope";23import * as Scope from "effect/Scope";
22import * as Semaphore from "effect/Semaphore";24import * as Semaphore from "effect/Semaphore";
23import * as Context from "effect/Context";25import * as Context from "effect/Context";
24import * as Stream from "effect/Stream";26import * as Stream from "effect/Stream";
27import * as Tracer from "effect/Tracer";
25import * as Reactivity from "effect/reactivity/Reactivity";28import * as Reactivity from "effect/reactivity/Reactivity";
26import * as Client from "effect/sql/SqlClient";29import * as Client from "effect/sql/SqlClient";
27import type { Connection } from "effect/sql/SqlConnection";30import type { Connection } from "effect/sql/SqlConnection";
...@@ -72,6 +75,33 @@ const ReadOnly = Context.Reference<boolean>("@t3tools/shared/nodeSqliteClient/Re...@@ -72,6 +75,33 @@ const ReadOnly = Context.Reference<boolean>("@t3tools/shared/nodeSqliteClient/Re
72export const readOnly = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>75export const readOnly = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
73 Effect.provideService(effect, ReadOnly, true);76 Effect.provideService(effect, ReadOnly, true);
7477
78/**
79 * Which connection ran the SQL, and how long it waited for that connection's
80 * lock. The wait is time queued behind other SQL, not time the event loop was
81 * blocked.
82 */
83interface Acquired {
84 readonly connection: "writer" | "reader";
85 readonly lockWaitMs: number;
86}
87
88const annotateAcquired = (fiber: Fiber.Fiber<unknown, unknown>, acquired: Acquired | undefined) => {
89 const span = Context.getOrUndefined(fiber.context, Tracer.ParentSpan);
90 if (span?._tag !== "Span" || acquired === undefined) return;
91 span.attribute("db.connection", acquired.connection);
92 span.attribute("db.lock_wait_ms", acquired.lockWaitMs);
93};
94
95/**
96 * Provided by the client's `withTransaction`. Effect acquires a transaction's
97 * connection before it makes the `sql.transaction` span current, so the acquire
98 * leaves its result here and the transaction body puts it on the span.
99 */
100const TransactionAcquire = Context.Reference<{ acquired?: Acquired } | undefined>(
101 "@t3tools/shared/nodeSqliteClient/TransactionAcquire",
102 { defaultValue: () => undefined },
103);
104
75export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError<UnsupportedNodeSqliteVersionError>()(105export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError<UnsupportedNodeSqliteVersionError>()(
76 "UnsupportedNodeSqliteVersionError",106 "UnsupportedNodeSqliteVersionError",
77 {107 {
...@@ -785,7 +815,7 @@ const make = Effect.fn("makeWithDatabase")(function* (...@@ -785,7 +815,7 @@ const make = Effect.fn("makeWithDatabase")(function* (
785 options.readerWorker && isFileDatabase ? yield* makeReader(options) : undefined;815 options.readerWorker && isFileDatabase ? yield* makeReader(options) : undefined;
786 const readerSemaphore = yield* Semaphore.make(1);816 const readerSemaphore = yield* Semaphore.make(1);
787817
788 const writer = { lock: semaphore, connection };818 const writer = { name: "writer" as const, lock: semaphore, connection };
789 // Picks the reader only for `readOnly` work, and the writer whenever the819 // Picks the reader only for `readOnly` work, and the writer whenever the
790 // reader is turned off.820 // reader is turned off.
791 const route = Effect.withFiber((fiber) =>821 const route = Effect.withFiber((fiber) =>
...@@ -793,28 +823,59 @@ const make = Effect.fn("makeWithDatabase")(function* (...@@ -793,28 +823,59 @@ const make = Effect.fn("makeWithDatabase")(function* (
793 ? Effect.map(acquireReader, (readerConnection) =>823 ? Effect.map(acquireReader, (readerConnection) =>
794 readerConnection === undefined824 readerConnection === undefined
795 ? writer825 ? writer
796 : { lock: readerSemaphore, connection: readerConnection },826 : { name: "reader" as const, lock: readerSemaphore, connection: readerConnection },
797 )827 )
798 : Effect.succeed(writer),828 : Effect.succeed(writer),
799 );829 );
800830
831 const acquired = (
832 target: { readonly name: Acquired["connection"] },
833 clock: Clock.Clock,
834 start: bigint,
835 ): Acquired => ({
836 connection: target.name,
837 lockWaitMs: Number((clock.monotonicTimeNanosUnsafe() - start) / 1000n) / 1000,
838 });
839
801 const acquirer = Effect.flatMap(route, (target) =>840 const acquirer = Effect.flatMap(route, (target) =>
802 target.lock.withPermits(1)(Effect.succeed(target.connection)),841 Effect.withFiber((fiber) => {
842 // The statement's span is current here only when it propagates, which
843 // Statement skips while tracing is off. Otherwise the current span is
844 // the caller's, so leave it alone.
845 if (
846 !fiber.getRef(References.TracerEnabled) ||
847 !fiber.getRef(Statement.SpanPropagationEnabled)
848 ) {
849 return target.lock.withPermits(1)(Effect.succeed(target.connection));
850 }
851 const clock = fiber.getRef(Clock.Clock);
852 const start = clock.monotonicTimeNanosUnsafe();
853 return target.lock.withPermits(1)(
854 Effect.sync(() => {
855 annotateAcquired(fiber, acquired(target, clock, start));
856 return target.connection;
857 }),
858 );
859 }),
803 );860 );
804 const transactionAcquirer = Effect.flatMap(route, (target) =>861 const transactionAcquirer = Effect.flatMap(route, (target) =>
805 Effect.uninterruptibleMask((restore) => {862 Effect.uninterruptibleMask((restore) => {
806 const fiber = Fiber.getCurrent()!;863 const fiber = Fiber.getCurrent()!;
807 const scope = Context.getUnsafe(fiber.context, Scope.Scope);864 const scope = Context.getUnsafe(fiber.context, Scope.Scope);
865 const clock = fiber.getRef(Clock.Clock);
866 const start = clock.monotonicTimeNanosUnsafe();
808 return Effect.as(867 return Effect.as(
809 Effect.tap(restore(target.lock.take(1)), () =>868 Effect.tap(restore(target.lock.take(1)), () => {
810 Scope.addFinalizer(scope, target.lock.release(1)),869 const pending = fiber.getRef(TransactionAcquire);
811 ),870 if (pending !== undefined) pending.acquired = acquired(target, clock, start);
871 return Scope.addFinalizer(scope, target.lock.release(1));
872 }),
812 target.connection,873 target.connection,
813 );874 );
814 }),875 }),
815 );876 );
816877
817 return yield* Client.make({878 const client = yield* Client.make({
818 acquirer,879 acquirer,
819 compiler,880 compiler,
820 transactionAcquirer,881 transactionAcquirer,
...@@ -831,6 +892,22 @@ const make = Effect.fn("makeWithDatabase")(function* (...@@ -831,6 +892,22 @@ const make = Effect.fn("makeWithDatabase")(function* (
831 ],892 ],
832 transformRows,893 transformRows,
833 });894 });
895 const withTransaction = client.withTransaction;
896 return Object.assign(client, {
897 // Each run gets its own holder, so a reused transaction Effect never
898 // carries an earlier run's attributes.
899 withTransaction: <A, E, R>(effect: Effect.Effect<A, E, R>) =>
900 Effect.suspend(() => {
901 const pending: { acquired?: Acquired } = {};
902 // The body runs with the `sql.transaction` span current. A nested
903 // transaction acquires nothing, so its span gets no attributes.
904 const body = Effect.withFiber<A, E, R>((fiber) => {
905 annotateAcquired(fiber, pending.acquired);
906 return effect;
907 });
908 return withTransaction(body).pipe(Effect.provideService(TransactionAcquire, pending));
909 }),
910 });
834});911});
835912
836export const layer = (config: SqliteClientConfig): Layer.Layer<Client.SqlClient, SqlError> =>913export const layer = (config: SqliteClientConfig): Layer.Layer<Client.SqlClient, SqlError> =>