From bac35a2ffcf90bb9fac7f63582bda9c549fd75ff Mon Sep 17 00:00:00 2001 From: Alex Southwell Date: Wed, 14 Oct 2026 19:03:15 +1100 Subject: [PATCH] perf(server): avoid quadratic resource history replay (#18825) Co-authored-by: SAPHID Co-authored-by: GPT-6.1 Sol --- apps/server/src/resourceTelemetry/Model.ts | 19 +++-- .../ResourceTelemetry.test.ts | 74 +++++++++++++++++++ .../ResourceTelemetryHistory.ts | 12 +-- 3 files changed, 92 insertions(+), 13 deletions(-) diff --git a/apps/server/src/resourceTelemetry/Model.ts b/apps/server/src/resourceTelemetry/Model.ts index 2582d7e9c17f78a0507cf1f9b0cde90fefba6a47..ccffd98757afbc7c2e6f9be522c843808ed2e0b8 100644 --- a/apps/server/src/resourceTelemetry/Model.ts +++ b/apps/server/src/resourceTelemetry/Model.ts @@ -52,6 +52,8 @@ export interface MergeProcessesInput { readonly previous: ReadonlyMap; readonly counters: TelemetryCounters; readonly updatePrevious: boolean; + /** History replay needs process deltas, but does not publish lifetime group counters. */ + readonly accumulateCounters?: boolean; } export interface MergeProcessesResult { @@ -574,14 +576,15 @@ export function mergeProcesses(input: MergeProcessesInput): MergeProcessesResult }); const ordered = orderProcessTree(normalized, rootPids); - const counters = input.updatePrevious - ? applyLifecycleCounters({ - counters: input.counters, - deltas: processDeltas, - current: nextPrevious, - previous: input.previous, - }) - : input.counters; + const counters = + input.updatePrevious && input.accumulateCounters !== false + ? applyLifecycleCounters({ + counters: input.counters, + deltas: processDeltas, + current: nextPrevious, + previous: input.previous, + }) + : input.counters; const backendProcesses = ordered.filter( (process) => categoryGroup(process.category) === "backend", ); diff --git a/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts b/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts index 5237704fc864083b1a98214326e58a42bfc225fe..c4273d7336aca103146016b0d103fd33448a4db5 100644 --- a/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts +++ b/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts @@ -142,6 +142,80 @@ function desktopSnapshot(sampledAtUnixMs: number): DesktopHostTelemetrySnapshot } describe("ResourceTelemetry", () => { + it.effect("replays churning history and retains baselines for returning processes", () => + Effect.gen(function* () { + const readAtMs = DateTime.toEpochMillis(yield* DateTime.now); + const sampleCount = 100; + const samples = Array.from({ length: sampleCount }, (_, index) => { + const sample = nativeSnapshot({ + sequence: index + 1, + sampledAtUnixMs: readAtMs - (sampleCount - index) * 1_000, + childCpuTimeMs: index * 10, + childWriteBytes: index * 100, + }); + const processes = sample.processes + .filter((entry) => entry.pid === process.pid || entry.pid === 4_242) + .map((entry) => + entry.pid === 4_242 + ? { ...entry, startTimeMs: index } + : { ...entry, ioWriteBytes: index * 100 }, + ); + if (index === 0 || index === 2) { + processes.push( + processSample({ + pid: 4_243, + ppid: process.pid, + startTimeMs: 500, + cpuTimeMs: index === 0 ? 100 : 350, + ioWriteBytes: index === 0 ? 1_000 : 5_000, + }), + ); + } + return { ...sample, processes }; + }); + const layerTelemetry = ResourceTelemetry.layer.pipe( + Layer.provide( + Layer.mergeAll( + NativeTelemetryClient.layerTest({ readHistory: () => Effect.succeed(samples) }), + DesktopTelemetryReceiver.layerTest(), + ResourceAttribution.layer, + ), + ), + ); + + yield* Effect.gen(function* () { + const telemetry = yield* ResourceTelemetry.ResourceTelemetry; + const history = yield* telemetry.readHistory({ + windowMs: sampleCount * 1_000, + bucketMs: 10_000, + }); + + expect(history.topProcesses).toHaveLength(sampleCount + 2); + expect( + history.topProcesses.find((entry) => entry.identity.pid === process.pid), + ).toMatchObject({ + sampleCount, + cpuTimeMs: (sampleCount - 1) * 10, + ioWriteBytes: (sampleCount - 1) * 100, + }); + const children = history.topProcesses.filter((entry) => entry.identity.pid === 4_242); + expect(children).toHaveLength(sampleCount); + for (const child of children) { + expect(child).toMatchObject({ sampleCount: 1, cpuTimeMs: 0, ioWriteBytes: 0 }); + } + expect(history.topProcesses.find((entry) => entry.identity.pid === 4_243)).toMatchObject({ + sampleCount: 2, + cpuTimeMs: 250, + ioWriteBytes: 4_000, + }); + expect(Math.max(...history.buckets.map((bucket) => bucket.maxProcessCount))).toBe(3); + expect(history.buckets.reduce((total, bucket) => total + bucket.ioWriteBytes, 0)).toBe( + (sampleCount - 1) * 100 + 4_000, + ); + }).pipe(Effect.provide(layerTelemetry)); + }), + ); + it.effect("enables live native and Electron collection only while changes are retained", () => Effect.gen(function* () { const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now); diff --git a/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.ts b/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.ts index 737a934de26d1621732a4af89befb9274851b488..890683ae8c7962ac07708bfa046a0a09e822e50c 100644 --- a/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.ts +++ b/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.ts @@ -15,7 +15,6 @@ import { mergeProcesses, processIdentityKey, type ProcessState, - type TelemetryCounters, } from "./Model.ts"; const MAX_HISTORY_WINDOW_MS = 60 * 60_000; @@ -167,8 +166,8 @@ export function buildResourceTelemetryHistory( const aggregateSamples: AggregateSample[] = []; const legacyBackendAggregateSamples: AggregateSample[] = []; const processSamples: ProcessSample[] = []; - let previous: ReadonlyMap = new Map(); - let counters: TelemetryCounters = emptyTelemetryCounters(); + const previous = new Map(); + const counters = emptyTelemetryCounters(); let previousSnapshotAtMs: number | undefined; for (const snapshot of snapshots) { @@ -216,9 +215,12 @@ export function buildResourceTelemetryHistory( previous, counters, updatePrevious: true, + accumulateCounters: false, }); - previous = new Map([...previous, ...merged.previous]); - counters = merged.counters; + // Keep absent identities' baselines without copying every exited process on each sample. + for (const [identityKey, state] of merged.previous) { + previous.set(identityKey, state); + } if (snapshot.sampledAtUnixMs < windowStartMs) { continue; } -- 2.54.0