feat(server): traces name what grew the heap, for ~0.3µs per span

apps/server/src/observability/HeapAttribution.test.ts

@@ -0,0 +1,69 @@
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Option from "effect/Option";
+import * as Context from "effect/Context";
+import * as TestClock from "effect/testing/TestClock";
+import * as Tracer from "effect/Tracer";
+
+import * as HeapAttribution from "./HeapAttribution.ts";
+
+const MB = 1024 * 1024;
+
+it.effect("names the span that grew the heap in a spike, not an older open or ended one", () =>
+  Effect.gen(function* () {
+    const ended: Array<Tracer.NativeSpan> = [];
+    const inner = Tracer.make({
+      span(options) {
+        const span = new Tracer.NativeSpan(options);
+        const end = span.end.bind(span);
+        span.end = (endTime, exit) => {
+          end(endTime, exit);
+          ended.push(span);
+        };
+        return span;
+      },
+    });
+    let heap = 300 * MB;
+    const attribution = yield* HeapAttribution.make(inner, () => heap);
+    yield* attribution.run;
+    // The sampler's baseline: heap before the spans allocate.
+    yield* TestClock.adjust("1 second");
+    const open = (name: string) =>
+      attribution.tracer.span({
+        name,
+        parent: Option.none(),
+        annotations: Context.empty(),
+        links: [],
+        startTime: 0n,
+        kind: "internal",
+        root: false,
+        sampled: true,
+      });
+
+    const subscription = open("ws.rpc.orchestration.subscribeShell");
+    const old = open("ProjectionStore.getSnapshot");
+    // Slow drift never spikes, but leaves the long-lived span far above its start.
+    for (let step = 0; step < 5; step++) {
+      heap += 80 * MB;
+      yield* TestClock.adjust("11 seconds");
+      if (step === 1) old.end(1n, undefined as never);
+    }
+    const small = open("ProjectionStore.getThreadShell");
+    const big = open("ProjectionStore.getThreadProjection");
+    heap += 2 * MB;
+    small.end(1n, undefined as never);
+    heap += 198 * MB;
+    yield* TestClock.adjust("2 seconds");
+
+    const spike = ended.find((span) => span.name === "server.heap.spike");
+    assert.isDefined(spike);
+    assert.deepEqual(spike!.attributes.get("suspects"), [
+      "ProjectionStore.getThreadProjection +200 MB",
+    ]);
+    big.end(2n, undefined as never);
+    subscription.end(3n, undefined as never);
+    assert.equal(big.attributes.get("memory.heap_growth_bytes"), 200 * MB);
+    assert.isUndefined(small.attributes.get("memory.heap_growth_bytes"));
+    assert.equal(old.attributes.get("memory.heap_growth_bytes"), 160 * MB);
+  }),
+);

apps/server/src/observability/HeapAttribution.ts

@@ -0,0 +1,108 @@
+// @effect-diagnostics nodeBuiltinImport:off - only node:v8 exposes a heap reading cheap enough to take per span.
+import * as NodeV8 from "node:v8";
+
+import * as Effect from "effect/Effect";
+import type * as Scope from "effect/Scope";
+import * as Tracer from "effect/Tracer";
+
+const MB = 1024 * 1024;
+// Smaller growth is noise from concurrent fibers; an attribute on every span would bloat exports.
+const SPAN_GROWTH_ATTRIBUTE_BYTES = 8 * MB;
+const SPIKE_GROWTH_BYTES = 128 * MB;
+const SPIKE_WINDOW_SAMPLES = 10;
+const SPIKE_COOLDOWN_SAMPLES = 60;
+const RECENT_SPANS = 32;
+const REPORTED_SPANS = 8;
+
+const readHeapUsed = () => NodeV8.getHeapStatistics().used_heap_size;
+
+interface Growth {
+  readonly name: string;
+  readonly growthBytes: number;
+}
+
+/**
+ * Heap growth is approximate: it includes other fibers' allocations and GC, so it names
+ * suspects, not culprits. A spike lists what was open or just grew.
+ */
+export const make = (tracer: Tracer.Tracer, heapUsed: () => number = readHeapUsed) =>
+  Effect.sync(() => {
+    const open = new Map<
+      Tracer.Span,
+      { readonly name: string; readonly startHeap: number; readonly startTick: number }
+    >();
+    const recent: Array<Growth & { readonly endTick: number }> = [];
+    const samples: Array<number> = [];
+    let cooldown = 0;
+    let tick = 0;
+
+    const wrapped = Tracer.make({
+      ...(tracer.context === undefined ? {} : { context: tracer.context }),
+      span(options) {
+        const span = tracer.span(options);
+        const startHeap = heapUsed();
+        open.set(span, { name: options.name, startHeap, startTick: tick });
+        const end = span.end.bind(span);
+        span.end = (endTime, exit) => {
+          open.delete(span);
+          const growthBytes = heapUsed() - startHeap;
+          if (growthBytes >= SPAN_GROWTH_ATTRIBUTE_BYTES) {
+            span.attribute("memory.heap_growth_bytes", growthBytes);
+            recent.push({ name: options.name, growthBytes, endTick: tick });
+            if (recent.length > RECENT_SPANS) recent.shift();
+          }
+          end(endTime, exit);
+        };
+        return span;
+      },
+    });
+
+    // Untraced: it runs every second and only emits a span when the heap spikes.
+    const sample = Effect.fnUntraced(function* () {
+      const heap = heapUsed();
+      tick++;
+      samples.push(heap);
+      if (samples.length > SPIKE_WINDOW_SAMPLES) samples.shift();
+      if (cooldown > 0) {
+        cooldown--;
+        return;
+      }
+      const growthBytes = heap - Math.min(...samples);
+      if (growthBytes < SPIKE_GROWTH_BYTES) return;
+      cooldown = SPIKE_COOLDOWN_SAMPLES;
+      const suspects = [
+        // Spans open since before the window, such as subscriptions, would claim all of its growth.
+        ...[...open.values()]
+          .filter(({ startTick }) => tick - startTick <= samples.length)
+          .map(({ name, startHeap }) => ({ name, growthBytes: heap - startHeap })),
+        ...recent.filter(({ endTick }) => tick - endTick <= samples.length),
+      ]
+        .filter((suspect) => suspect.growthBytes >= SPAN_GROWTH_ATTRIBUTE_BYTES)
+        .toSorted((left, right) => right.growthBytes - left.growthBytes)
+        .slice(0, REPORTED_SPANS)
+        .map(({ name, growthBytes }) => `${name} +${Math.round(growthBytes / MB)} MB`);
+      recent.length = 0;
+      // A span, not a log: this fiber starts before the server logger is provided.
+      yield* Effect.void.pipe(
+        Effect.withSpan("server.heap.spike", {
+          root: true,
+          level: "Warn",
+          attributes: {
+            heapUsedMb: Math.round(heap / MB),
+            growthMb: Math.round(growthBytes / MB),
+            suspects,
+          },
+        }),
+      );
+    });
+
+    const run: Effect.Effect<void, never, Scope.Scope> = sample().pipe(
+      Effect.andThen(Effect.sleep("1 second")),
+      Effect.forever,
+      Effect.withTracer(wrapped),
+      Effect.forkScoped,
+      Effect.asVoid,
+    );
+
+    return { tracer: wrapped, run };
+  });

apps/server/src/observability/Observability.ts

@@ -16,6 +16,7 @@ import * as ServerConfig from "../config.ts";
 import * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts";
 import * as ServerLogger from "../serverLogger.ts";
 import * as BrowserTraceCollector from "./BrowserTraceCollector.ts";
+import * as HeapAttribution from "./HeapAttribution.ts";
 
 export const layer = Layer.unwrap(
   Effect.gen(function* () {
@@ -71,8 +72,11 @@ export const layer = Layer.unwrap(
           ...(delegate ? { delegate } : {}),
         });
 
+        const heapAttribution = yield* HeapAttribution.make(tracer);
+        yield* heapAttribution.run;
+
         return Layer.mergeAll(
-          Layer.succeed(Tracer.Tracer, tracer),
+          Layer.succeed(Tracer.Tracer, heapAttribution.tracer),
           BrowserTraceCollector.layer(sink),
         );
       }),