| author |  Alex Southwell <saphid@gmail.com> 2026-10-14 19:03:15+11:00 |
| committer |  GitHub <noreply@github.com> 2026-10-14 01:03:15-07:00 |
| log | bac35a2ffcf90bb9fac7f63582bda9c549fd75ff |
| tree | 0d7c74c949b46177091aa22e49c7bc4fce883bb0 |
| parent | 8855a6dd052a8cbc9aac125dc2e8d14a0c2decd6 |
| signature | Signed by PGP key B5690EEEBB952194 |
perf(server): avoid quadratic resource history replay (#18825)
Co-authored-by: SAPHID <saphid@users.noreply.github.com>
Co-authored-by: GPT-6.1 Sol <noreply@openai.com>3 files changed, 92 insertions(+), 13 deletions(-)
apps
server
src
resourceTelemetry
apps/server/src/resourceTelemetry/Model.ts+11-8
| ... | ... | @@ -52,6 +52,8 @@ export interface MergeProcessesInput { |
| 52 | 52 | readonly previous: ReadonlyMap<string, ProcessState>; |
| 53 | 53 | readonly counters: TelemetryCounters; |
| 54 | 54 | readonly updatePrevious: boolean; |
| 55 | /** History replay needs process deltas, but does not publish lifetime group counters. */ | |
| 56 | readonly accumulateCounters?: boolean; | |
| 55 | 57 | } |
| 56 | 58 | |
| 57 | 59 | export interface MergeProcessesResult { |
| ... | ... | @@ -574,14 +576,15 @@ export function mergeProcesses(input: MergeProcessesInput): MergeProcessesResult |
| 574 | 576 | }); |
| 575 | 577 | const ordered = orderProcessTree(normalized, rootPids); |
| 576 | 578 | |
| 577 | const counters = input.updatePrevious | |
| 578 | ? applyLifecycleCounters({ | |
| 579 | counters: input.counters, | |
| 580 | deltas: processDeltas, | |
| 581 | current: nextPrevious, | |
| 582 | previous: input.previous, | |
| 583 | }) | |
| 584 | : input.counters; | |
| 579 | const counters = | |
| 580 | input.updatePrevious && input.accumulateCounters !== false | |
| 581 | ? applyLifecycleCounters({ | |
| 582 | counters: input.counters, | |
| 583 | deltas: processDeltas, | |
| 584 | current: nextPrevious, | |
| 585 | previous: input.previous, | |
| 586 | }) | |
| 587 | : input.counters; | |
| 585 | 588 | const backendProcesses = ordered.filter( |
| 586 | 589 | (process) => categoryGroup(process.category) === "backend", |
| 587 | 590 | ); |
apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts+74
| ... | ... | @@ -142,6 +142,80 @@ function desktopSnapshot(sampledAtUnixMs: number): DesktopHostTelemetrySnapshot |
| 142 | 142 | } |
| 143 | 143 | |
| 144 | 144 | describe("ResourceTelemetry", () => { |
| 145 | it.effect("replays churning history and retains baselines for returning processes", () => | |
| 146 | Effect.gen(function* () { | |
| 147 | const readAtMs = DateTime.toEpochMillis(yield* DateTime.now); | |
| 148 | const sampleCount = 100; | |
| 149 | const samples = Array.from({ length: sampleCount }, (_, index) => { | |
| 150 | const sample = nativeSnapshot({ | |
| 151 | sequence: index + 1, | |
| 152 | sampledAtUnixMs: readAtMs - (sampleCount - index) * 1_000, | |
| 153 | childCpuTimeMs: index * 10, | |
| 154 | childWriteBytes: index * 100, | |
| 155 | }); | |
| 156 | const processes = sample.processes | |
| 157 | .filter((entry) => entry.pid === process.pid || entry.pid === 4_242) | |
| 158 | .map((entry) => | |
| 159 | entry.pid === 4_242 | |
| 160 | ? { ...entry, startTimeMs: index } | |
| 161 | : { ...entry, ioWriteBytes: index * 100 }, | |
| 162 | ); | |
| 163 | if (index === 0 || index === 2) { | |
| 164 | processes.push( | |
| 165 | processSample({ | |
| 166 | pid: 4_243, | |
| 167 | ppid: process.pid, | |
| 168 | startTimeMs: 500, | |
| 169 | cpuTimeMs: index === 0 ? 100 : 350, | |
| 170 | ioWriteBytes: index === 0 ? 1_000 : 5_000, | |
| 171 | }), | |
| 172 | ); | |
| 173 | } | |
| 174 | return { ...sample, processes }; | |
| 175 | }); | |
| 176 | const layerTelemetry = ResourceTelemetry.layer.pipe( | |
| 177 | Layer.provide( | |
| 178 | Layer.mergeAll( | |
| 179 | NativeTelemetryClient.layerTest({ readHistory: () => Effect.succeed(samples) }), | |
| 180 | DesktopTelemetryReceiver.layerTest(), | |
| 181 | ResourceAttribution.layer, | |
| 182 | ), | |
| 183 | ), | |
| 184 | ); | |
| 185 | ||
| 186 | yield* Effect.gen(function* () { | |
| 187 | const telemetry = yield* ResourceTelemetry.ResourceTelemetry; | |
| 188 | const history = yield* telemetry.readHistory({ | |
| 189 | windowMs: sampleCount * 1_000, | |
| 190 | bucketMs: 10_000, | |
| 191 | }); | |
| 192 | ||
| 193 | expect(history.topProcesses).toHaveLength(sampleCount + 2); | |
| 194 | expect( | |
| 195 | history.topProcesses.find((entry) => entry.identity.pid === process.pid), | |
| 196 | ).toMatchObject({ | |
| 197 | sampleCount, | |
| 198 | cpuTimeMs: (sampleCount - 1) * 10, | |
| 199 | ioWriteBytes: (sampleCount - 1) * 100, | |
| 200 | }); | |
| 201 | const children = history.topProcesses.filter((entry) => entry.identity.pid === 4_242); | |
| 202 | expect(children).toHaveLength(sampleCount); | |
| 203 | for (const child of children) { | |
| 204 | expect(child).toMatchObject({ sampleCount: 1, cpuTimeMs: 0, ioWriteBytes: 0 }); | |
| 205 | } | |
| 206 | expect(history.topProcesses.find((entry) => entry.identity.pid === 4_243)).toMatchObject({ | |
| 207 | sampleCount: 2, | |
| 208 | cpuTimeMs: 250, | |
| 209 | ioWriteBytes: 4_000, | |
| 210 | }); | |
| 211 | expect(Math.max(...history.buckets.map((bucket) => bucket.maxProcessCount))).toBe(3); | |
| 212 | expect(history.buckets.reduce((total, bucket) => total + bucket.ioWriteBytes, 0)).toBe( | |
| 213 | (sampleCount - 1) * 100 + 4_000, | |
| 214 | ); | |
| 215 | }).pipe(Effect.provide(layerTelemetry)); | |
| 216 | }), | |
| 217 | ); | |
| 218 | ||
| 145 | 219 | it.effect("enables live native and Electron collection only while changes are retained", () => |
| 146 | 220 | Effect.gen(function* () { |
| 147 | 221 | const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now); |
apps/server/src/resourceTelemetry/ResourceTelemetryHistory.ts+7-5
| ... | ... | @@ -15,7 +15,6 @@ import { |
| 15 | 15 | mergeProcesses, |
| 16 | 16 | processIdentityKey, |
| 17 | 17 | type ProcessState, |
| 18 | type TelemetryCounters, | |
| 19 | 18 | } from "./Model.ts"; |
| 20 | 19 | |
| 21 | 20 | const MAX_HISTORY_WINDOW_MS = 60 * 60_000; |
| ... | ... | @@ -167,8 +166,8 @@ export function buildResourceTelemetryHistory( |
| 167 | 166 | const aggregateSamples: AggregateSample[] = []; |
| 168 | 167 | const legacyBackendAggregateSamples: AggregateSample[] = []; |
| 169 | 168 | const processSamples: ProcessSample[] = []; |
| 170 | let previous: ReadonlyMap<string, ProcessState> = new Map(); | |
| 171 | let counters: TelemetryCounters = emptyTelemetryCounters(); | |
| 169 | const previous = new Map<string, ProcessState>(); | |
| 170 | const counters = emptyTelemetryCounters(); | |
| 172 | 171 | let previousSnapshotAtMs: number | undefined; |
| 173 | 172 | |
| 174 | 173 | for (const snapshot of snapshots) { |
| ... | ... | @@ -216,9 +215,12 @@ export function buildResourceTelemetryHistory( |
| 216 | 215 | previous, |
| 217 | 216 | counters, |
| 218 | 217 | updatePrevious: true, |
| 218 | accumulateCounters: false, | |
| 219 | 219 | }); |
| 220 | previous = new Map([...previous, ...merged.previous]); | |
| 221 | counters = merged.counters; | |
| 220 | // Keep absent identities' baselines without copying every exited process on each sample. | |
| 221 | for (const [identityKey, state] of merged.previous) { | |
| 222 | previous.set(identityKey, state); | |
| 223 | } | |
| 222 | 224 | if (snapshot.sampledAtUnixMs < windowStartMs) { |
| 223 | 225 | continue; |
| 224 | 226 | } |