fix(server): log why websocket connections close#24285

@@ -61,8 +104,44 @@ export interface EventLoopReadings {
   /** One-minute load average, undefined on Windows. */
   readonly loadAverage: number | undefined;
   readonly cpuCount: number;
+  /** The main thread's own counters over the readings, Linux only. */
+  readonly mainThread: MainThreadCounters | undefined;
 }
 
+/** CPU and runqueue time in ns, storage bytes the thread caused to be read. */
+export interface MainThreadCounters {
+  readonly cpuNs: number;
+  readonly runqueueNs: number;
+  readonly readBytes: number;
+  readonly readSyscalls: number;
+}
+
+// process.resourceUsage() sums every thread, including the SQLite worker. A synchronous
+// read of thread-self runs on the main thread, so it names that thread. Procfs files are
+// generated in memory, never read from disk.
+const readMainThreadCounters = (): MainThreadCounters | undefined => {
+  try {
+    const [cpuNs = Number.NaN, runqueueNs = Number.NaN] = NodeFS.readFileSync(
+      "/proc/thread-self/schedstat",
+      "utf8",
+    )
+      .trim()
+      .split(" ")
+      .map(Number);

@@ -131,6 +133,29 @@ waiting. Read CPU together with page faults:
   file writes.
 - Many `involuntaryContextSwitches` mean other processes were competing for the CPU.
 
+On Linux, the `mainThread*` fields cover only the main thread, not the SQLite worker.
+`mainThreadBlockedMs` is busy time spent neither on a CPU nor waiting for one: a wait in the kernel.
+With a large `mainThreadReadMb`, that wait was the main thread reading from disk. Both cover the
+window, not just the stall.

@@ -25,11 +28,51 @@ import * as Metrics from "./Metrics.ts";
 const RESOLUTION_MS = 1000;
 const STALL_THRESHOLD_MS = 2000;
 const SAMPLE_INTERVAL = "30 seconds";
-const STALL_LOG_MAX_BYTES = 256 * 1024;
+// A line with candidates is about 1.5 KB, so a file holds about 1,000 stalls and the
+// rotated backup keeps the documented last 1000 or more.
+const STALL_LOG_MAX_BYTES = 1536 * 1024;
 // Long enough that users felt it. Shorter stalls stay in the log only.
 const NOTICEABLE_STALL_MS = 5000;
 // Clients show what they get, so an old stall must not explain a new drop.
 const NOTICEABLE_STALL_MAX_AGE_MS = 5 * 60_000;
+// A span that blocked the loop for a stall lasted at least the stall, and stalls start
+// above STALL_THRESHOLD_MS, so shorter spans can't be the cause.
+export const SLOW_SPAN_MIN_MS = STALL_THRESHOLD_MS - RESOLUTION_MS;
+// Long subscriptions and network calls also end up here. 128 holds a few sample
+// windows of them on a busy host, so the span that blocked is still there at the tick.
+const SLOW_SPAN_RING_SIZE = 128;
+const STALL_CANDIDATE_LIMIT = 3;
+const CANDIDATE_ATTRIBUTE_KEYS = [
+  "rpc.method",
+  "db.connection",
+  "db.lock_wait_ms",
+  "effect.sql.method",
+  "db.query.text",

@@ -0,0 +1,108 @@
+import * as Cause from "effect/Cause";
+import * as Clock from "effect/Clock";
+import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+import * as Option from "effect/Option";
+import type * as HttpServerRequest from "effect/http/HttpServerRequest";
+import * as Socket from "effect/socket/Socket";
+
+// RFC 6455 caps a close reason at 123 bytes; anything longer came from a broken peer.
+const CLOSE_REASON_MAX_CHARS = 123;
+const ADDRESS_MAX_CHARS = 64;
+
+/** What the server saw of one `/ws` connection before it ended. */
+export interface WsConnectionObservation {
+  closeCode: number | undefined;
+  closeReason: string | undefined;
+  lastInboundAtMs: number | undefined;
+  invalidated: boolean;
+}
+
+export const makeObservation = (): WsConnectionObservation => ({
+  closeCode: undefined,
+  closeReason: undefined,
+  lastInboundAtMs: undefined,
+  invalidated: false,
+});
+
+const observeSocket = (socket: Socket.Socket, observation: WsConnectionObservation) =>
+  Socket.make({
+    reader: Effect.map(socket.reader, (reader) => ({
+      ...reader,
+      pull: reader.pull.pipe(
+        Effect.tap(() =>
+          Clock.currentTimeMillis.pipe(
+            Effect.map((now) => {
+              observation.lastInboundAtMs = now;
+            }),
+          ),
+        ),
+        Effect.tapError((error) =>
+          Effect.sync(() => {
+            if (error.reason._tag !== "SocketCloseError") return;
+            observation.closeCode = error.reason.code;
+            observation.closeReason = error.reason.closeReason?.slice(0, CLOSE_REASON_MAX_CHARS);
+          }),
+        ),
+      ),
+    })),
+    writer: socket.writer,
+  });
+
+/**
+ * Effect's RPC server catches the close error its socket reader raises, so the close
+ * code exists only inside the reader. This request upgrades to a socket that keeps it.
+ */
+export const observeUpgrade = (
+  request: HttpServerRequest.HttpServerRequest,
+  observation: WsConnectionObservation,
+): HttpServerRequest.HttpServerRequest =>
+  new Proxy(request, {
+    get: (target, key) =>
+      key === "upgrade"
+        ? Effect.map(target.upgrade, (socket) => observeSocket(socket, observation))
+        : Reflect.get(target, key, target),
+  });
+
+/**
+ * 1006 means the TCP connection ended with no close frame: a network drop, a killed
+ * client, or a proxy that gave up. Any other code came in a close frame.
+ */
+export const wsCloseAttributes = (input: {
+  readonly observation: WsConnectionObservation;
+  readonly exit: Exit.Exit<unknown, unknown>;
+  readonly connectedAtMs: number;
+  readonly closedAtMs: number;
+}) => {
+  const { observation, exit, connectedAtMs, closedAtMs } = input;
+  const endedBy = observation.invalidated
+    ? "session_invalidated"
+    : observation.closeCode === 1006
+      ? "connection_lost"
+      : observation.closeCode !== undefined
+        ? "close_frame"
+        : Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)
+          ? "server_interrupted"
+          : Exit.isFailure(exit)
+            ? "server_error"
+            : "unknown";
+  return {
+    endedBy,
+    ...(observation.closeCode === undefined ? {} : { closeCode: observation.closeCode }),
+    ...(observation.closeReason ? { closeReason: observation.closeReason } : {}),