From 922c17f2425e7539f6d0e5480ac59599b13a4240 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 29 Oct 2026 10:53:14 -0700 Subject: [PATCH 1/3] feat(server): traces name what grew the heap Prometheus showed server heap jumping from ~280 MB to 1-5 GB between 30-second samples, but no span carried memory, so nothing could say what allocated. The server tracer now reads V8 heap at span start and end (~0.3 us each) and tags spans that grew it by 8 MB or more with memory.heap_growth_bytes. A 1-second sampler emits a server.heap.spike warning span when heap grows 128 MB within 10 s, listing the open or just-finished spans that grew it most. Co-Authored-By: Claude Opus 5.5 --- .../src/observability/HeapAttribution.test.ts | 59 ++++++++++ .../src/observability/HeapAttribution.ts | 103 ++++++++++++++++++ .../server/src/observability/Observability.ts | 6 +- 3 files changed, 167 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/observability/HeapAttribution.test.ts create mode 100644 apps/server/src/observability/HeapAttribution.ts diff --git a/apps/server/src/observability/HeapAttribution.test.ts b/apps/server/src/observability/HeapAttribution.test.ts new file mode 100644 index 0000000000000..17eed43d374ce --- /dev/null +++ b/apps/server/src/observability/HeapAttribution.test.ts @@ -0,0 +1,59 @@ +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 open span that grew the heap in a spike, and tags large growth", () => + Effect.gen(function* () { + const ended: Array = []; + 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 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); + assert.equal(big.attributes.get("memory.heap_growth_bytes"), 200 * MB); + assert.isUndefined(small.attributes.get("memory.heap_growth_bytes")); + }), +); diff --git a/apps/server/src/observability/HeapAttribution.ts b/apps/server/src/observability/HeapAttribution.ts new file mode 100644 index 0000000000000..1cf008f39d2ba --- /dev/null +++ b/apps/server/src/observability/HeapAttribution.ts @@ -0,0 +1,103 @@ +// @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 = Effect.fnUntraced(function* ( + tracer: Tracer.Tracer, + heapUsed: () => number = readHeapUsed, +) { + const open = new Map(); + const recent: Array = []; + const samples: Array = []; + let cooldown = 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 }); + 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 }); + 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(); + 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 = [ + ...[...open.values()].map(({ name, startHeap }) => ({ name, growthBytes: heap - startHeap })), + ...recent, + ] + .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; + yield* Effect.logWarning( + `heap grew ${Math.round(growthBytes / MB)} MB in ${samples.length} s`, + ).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 = sample().pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.forever, + Effect.withTracer(wrapped), + Effect.forkScoped, + Effect.asVoid, + ); + + return { tracer: wrapped, run }; +}); diff --git a/apps/server/src/observability/Observability.ts b/apps/server/src/observability/Observability.ts index 343851e3d1d11..166d8ac194a8c 100644 --- a/apps/server/src/observability/Observability.ts +++ b/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), ); }), From 86133aceb0246c1086515d5728f9557114f85981 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 29 Oct 2026 11:28:23 -0700 Subject: [PATCH 2/3] fix(server): heap spikes no longer blame long-lived subscription spans An open span reported all heap growth since it started, so a subscription open for hours outranked the request that actually spiked and could push it out of the suspect list. Open spans count only if they started inside the spike window. The spike is now a span only: its log line ran before the server logger was provided and skipped OTLP logs. Co-Authored-By: Claude Opus 5.5 --- .../src/observability/HeapAttribution.test.ts | 9 +- .../src/observability/HeapAttribution.ts | 149 +++++++++--------- 2 files changed, 85 insertions(+), 73 deletions(-) diff --git a/apps/server/src/observability/HeapAttribution.test.ts b/apps/server/src/observability/HeapAttribution.test.ts index 17eed43d374ce..03f410010cd5d 100644 --- a/apps/server/src/observability/HeapAttribution.test.ts +++ b/apps/server/src/observability/HeapAttribution.test.ts @@ -9,7 +9,7 @@ import * as HeapAttribution from "./HeapAttribution.ts"; const MB = 1024 * 1024; -it.effect("names the open span that grew the heap in a spike, and tags large growth", () => +it.effect("names the span that grew the heap in a spike, not an older open one", () => Effect.gen(function* () { const ended: Array = []; const inner = Tracer.make({ @@ -40,6 +40,12 @@ it.effect("names the open span that grew the heap in a spike, and tags large gro sampled: true, }); + const subscription = open("ws.rpc.orchestration.subscribeShell"); + // 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"); + } const small = open("ProjectionStore.getThreadShell"); const big = open("ProjectionStore.getThreadProjection"); heap += 2 * MB; @@ -53,6 +59,7 @@ it.effect("names the open span that grew the heap in a spike, and tags large gro "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")); }), diff --git a/apps/server/src/observability/HeapAttribution.ts b/apps/server/src/observability/HeapAttribution.ts index 1cf008f39d2ba..43fb215b91c29 100644 --- a/apps/server/src/observability/HeapAttribution.ts +++ b/apps/server/src/observability/HeapAttribution.ts @@ -25,79 +25,84 @@ interface Growth { * 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 = Effect.fnUntraced(function* ( - tracer: Tracer.Tracer, - heapUsed: () => number = readHeapUsed, -) { - const open = new Map(); - const recent: Array = []; - const samples: Array = []; - let cooldown = 0; +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 = []; + const samples: Array = []; + 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 }); - 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 }); - if (recent.length > RECENT_SPANS) recent.shift(); - } - end(endTime, exit); - }; - return span; - }, - }); + 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 }); + 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(); - 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 = [ - ...[...open.values()].map(({ name, startHeap }) => ({ name, growthBytes: heap - startHeap })), - ...recent, - ] - .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; - yield* Effect.logWarning( - `heap grew ${Math.round(growthBytes / MB)} MB in ${samples.length} s`, - ).pipe( - Effect.withSpan("server.heap.spike", { - root: true, - level: "Warn", - attributes: { - heapUsedMb: Math.round(heap / MB), - growthMb: Math.round(growthBytes / MB), - suspects, - }, - }), - ); - }); + // 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((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 = sample().pipe( - Effect.andThen(Effect.sleep("1 second")), - Effect.forever, - Effect.withTracer(wrapped), - Effect.forkScoped, - Effect.asVoid, - ); + const run: Effect.Effect = sample().pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.forever, + Effect.withTracer(wrapped), + Effect.forkScoped, + Effect.asVoid, + ); - return { tracer: wrapped, run }; -}); + return { tracer: wrapped, run }; + }); From 06554ac6efd5fd436153c4930622558f7dd94a1e Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 29 Oct 2026 13:18:38 -0700 Subject: [PATCH 3/3] fix(server): heap spikes no longer blame spans that ended long before Co-Authored-By: Claude Opus 5.5 --- apps/server/src/observability/HeapAttribution.test.ts | 5 ++++- apps/server/src/observability/HeapAttribution.ts | 6 +++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/apps/server/src/observability/HeapAttribution.test.ts b/apps/server/src/observability/HeapAttribution.test.ts index 03f410010cd5d..b355ba5c35e1f 100644 --- a/apps/server/src/observability/HeapAttribution.test.ts +++ b/apps/server/src/observability/HeapAttribution.test.ts @@ -9,7 +9,7 @@ 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 one", () => +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 = []; const inner = Tracer.make({ @@ -41,10 +41,12 @@ it.effect("names the span that grew the heap in a spike, not an older open one", }); 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"); @@ -62,5 +64,6 @@ it.effect("names the span that grew the heap in a spike, not an older open one", 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); }), ); diff --git a/apps/server/src/observability/HeapAttribution.ts b/apps/server/src/observability/HeapAttribution.ts index 43fb215b91c29..4b18efd0fb911 100644 --- a/apps/server/src/observability/HeapAttribution.ts +++ b/apps/server/src/observability/HeapAttribution.ts @@ -31,7 +31,7 @@ export const make = (tracer: Tracer.Tracer, heapUsed: () => number = readHeapUse Tracer.Span, { readonly name: string; readonly startHeap: number; readonly startTick: number } >(); - const recent: Array = []; + const recent: Array = []; const samples: Array = []; let cooldown = 0; let tick = 0; @@ -48,7 +48,7 @@ export const make = (tracer: Tracer.Tracer, heapUsed: () => number = readHeapUse const growthBytes = heapUsed() - startHeap; if (growthBytes >= SPAN_GROWTH_ATTRIBUTE_BYTES) { span.attribute("memory.heap_growth_bytes", growthBytes); - recent.push({ name: options.name, growthBytes }); + recent.push({ name: options.name, growthBytes, endTick: tick }); if (recent.length > RECENT_SPANS) recent.shift(); } end(endTime, exit); @@ -75,7 +75,7 @@ export const make = (tracer: Tracer.Tracer, heapUsed: () => number = readHeapUse ...[...open.values()] .filter(({ startTick }) => tick - startTick <= samples.length) .map(({ name, startHeap }) => ({ name, growthBytes: heap - startHeap })), - ...recent, + ...recent.filter(({ endTick }) => tick - endTick <= samples.length), ] .filter((suspect) => suspect.growthBytes >= SPAN_GROWTH_ATTRIBUTE_BYTES) .toSorted((left, right) => right.growthBytes - left.growthBytes)