Open History to see the last commit
1import type {
2 DesktopElectronProcessMetric,
3 DesktopHostTelemetrySnapshot,
4 ResourceMonitorProcessSample,
5 ResourceMonitorSnapshotEvent,
6 ResourceTelemetryAggregate,
7 ResourceTelemetryProcess,
8 ResourceTelemetryProcessCategory,
9} from "@t3tools/contracts";
10import * as DateTime from "effect/DateTime";
11import * as Option from "effect/Option";
12
13const MAX_DELTA_INTERVAL_MS = 30_000;
14const ELECTRON_IDENTITY_TOLERANCE_MS = 2_000;
15
16export interface ProcessState {
17 readonly process: ResourceTelemetryProcess;
18 readonly sampledAtMs: number;
19}
20
21export interface GroupCounters {
22 readonly cpuTimeMs: number;
23 readonly ioReadBytes: number;
24 readonly ioWriteBytes: number;
25 readonly processStarts: number;
26 readonly processExits: number;
27}
28
29export interface TelemetryCounters {
30 readonly backend: GroupCounters;
31 readonly electron: GroupCounters;
32 readonly monitor: GroupCounters;
33 readonly allT3: GroupCounters;
34}
35
36export interface ProcessDelta {
37 readonly identityKey: string;
38 readonly category: ResourceTelemetryProcessCategory;
39 readonly cpuTimeMs: number;
40 readonly ioReadBytes: number;
41 readonly ioWriteBytes: number;
42}
43
44export interface MergeProcessesInput {
45 readonly serverPid: number;
46 readonly sidecarPid: Option.Option<number>;
47 readonly fallbackSampledAtMs: number;
48 readonly nativeSnapshot: Option.Option<ResourceMonitorSnapshotEvent>;
49 readonly desktopSnapshot: Option.Option<DesktopHostTelemetrySnapshot>;
50 readonly electronRootPids?: ReadonlySet<number>;
51 readonly electronRootStartTimes?: ReadonlyMap<number, number>;
52 readonly previous: ReadonlyMap<string, ProcessState>;
53 readonly counters: TelemetryCounters;
54 readonly updatePrevious: boolean;
55 readonly accumulateCounters?: boolean;
56}
57
58export interface MergeProcessesResult {
59 readonly sampledAtMs: number;
60 readonly processes: ReadonlyArray<ResourceTelemetryProcess>;
61 readonly previous: ReadonlyMap<string, ProcessState>;
62 readonly counters: TelemetryCounters;
63 readonly groups: {
64 readonly backend: ResourceTelemetryAggregate;
65 readonly electron: ResourceTelemetryAggregate;
66 readonly monitor: ResourceTelemetryAggregate;
67 readonly allT3: ResourceTelemetryAggregate;
68 };
69 readonly deltas: ReadonlyArray<ProcessDelta>;
70}
71
72const emptyGroupCounters = (): GroupCounters => ({
73 cpuTimeMs: 0,
74 ioReadBytes: 0,
75 ioWriteBytes: 0,
76 processStarts: 0,
77 processExits: 0,
78});
79
80export const emptyTelemetryCounters = (): TelemetryCounters => ({
81 backend: emptyGroupCounters(),
82 electron: emptyGroupCounters(),
83 monitor: emptyGroupCounters(),
84 allT3: emptyGroupCounters(),
85});
86
87export function processIdentityKey(pid: number, startTimeMs: number): string {
88 return `${pid}:${startTimeMs}`;
89}
90
91function finiteNonNegative(value: number): number {
92 return Number.isFinite(value) ? Math.max(0, value) : 0;
93}
94
95function categoryGroup(
96 category: ResourceTelemetryProcessCategory,
97): "backend" | "electron" | "monitor" {
98 if (category === "resource-monitor") return "monitor";
99 if (category.startsWith("electron-")) return "electron";
100 return "backend";
101}
102
103function electronCategory(metric: DesktopElectronProcessMetric): ResourceTelemetryProcessCategory {
104 switch (metric.type) {
105 case "Browser":
106 return "electron-main";
107 case "Tab":
108 return "electron-renderer";
109 case "GPU":
110 return "electron-gpu";
111 default:
112 return "electron-utility";
113 }
114}
115
116function inferredElectronCategory(
117 process: ResourceMonitorProcessSample,
118): ResourceTelemetryProcessCategory {
119 const command = process.command.toLowerCase();
120 if (command.includes("--type=renderer")) return "electron-renderer";
121 if (command.includes("--type=gpu-process")) return "electron-gpu";
122 return "electron-utility";
123}
124
125function matchElectronMetric(
126 process: ResourceMonitorProcessSample,
127 metricsByPid: ReadonlyMap<number, DesktopElectronProcessMetric>,
128): DesktopElectronProcessMetric | undefined {
129 const metric = metricsByPid.get(process.pid);
130 if (!metric) return undefined;
131 return Math.abs(metric.creationTimeMs - process.startTimeMs) <= ELECTRON_IDENTITY_TOLERANCE_MS
132 ? metric
133 : undefined;
134}
135
136function syntheticNativeSample(
137 metric: DesktopElectronProcessMetric,
138 sampledAtMs: number,
139 previous: ProcessState | undefined,
140): ResourceMonitorProcessSample {
141 const cpuTimeMs =
142 metric.cumulativeCpuSeconds !== undefined
143 ? Math.max(0, Math.round(metric.cumulativeCpuSeconds * 1_000))
144 : previous
145 ? previous.process.cpuTimeMs +
146 Math.max(0, ((sampledAtMs - previous.sampledAtMs) * metric.cpuPercent) / 100)
147 : 0;
148 return {
149 pid: metric.pid,
150 ppid: 0,
151 startTimeMs: metric.creationTimeMs,
152 runTimeMs: Math.max(0, sampledAtMs - metric.creationTimeMs),
153 name: metric.name ?? metric.serviceName ?? metric.type,
154 command: metric.name ?? metric.serviceName ?? metric.type,
155 status: "Running",
156 cpuPercent: metric.cpuPercent,
157 cpuTimeMs,
158 residentBytes: metric.workingSetBytes,
159 virtualBytes: 0,
160 ioReadBytes: 0,
161 ioWriteBytes: 0,
162 ioSemantics: "storage",
163 };
164}
165
166function processDepths(
167 processes: ReadonlyArray<ResourceMonitorProcessSample>,
168 roots: ReadonlySet<number>,
169): ReadonlyMap<number, number> {
170 const childrenByParent = new Map<number, number[]>();
171 for (const process of processes) {
172 const children = childrenByParent.get(process.ppid) ?? [];
173 children.push(process.pid);
174 childrenByParent.set(process.ppid, children);
175 }
176
177 const depths = new Map<number, number>();
178 const queue = [...roots].map((pid) => ({ pid, depth: 0 }));
179 while (queue.length > 0) {
180 const current = queue.shift();
181 if (!current || depths.has(current.pid)) continue;
182 depths.set(current.pid, current.depth);
183 for (const childPid of childrenByParent.get(current.pid) ?? []) {
184 queue.push({ pid: childPid, depth: current.depth + 1 });
185 }
186 }
187 return depths;
188}
189
190function isElectronDescendant(
191 pid: number,
192 processesByPid: ReadonlyMap<number, ResourceMonitorProcessSample>,
193 electronPids: ReadonlySet<number>,
194 serverPid: number,
195): boolean {
196 const visited = new Set<number>();
197 let currentPid = pid;
198 while (!visited.has(currentPid)) {
199 visited.add(currentPid);
200 // The backend's children belong to the backend even when Electron launched it.
201 if (currentPid === serverPid) return false;
202 if (electronPids.has(currentPid)) return true;
203 const current = processesByPid.get(currentPid);
204 if (!current || current.ppid <= 0 || current.ppid === currentPid) return false;
205 currentPid = current.ppid;
206 }
207 return false;
208}
209
210function hasElectronAncestor(
211 process: ResourceMonitorProcessSample,
212 processesByPid: ReadonlyMap<number, ResourceMonitorProcessSample>,
213 electronPids: ReadonlySet<number>,
214): boolean {
215 const visited = new Set<number>();
216 let currentPid = process.ppid;
217 while (currentPid > 0 && !visited.has(currentPid)) {
218 visited.add(currentPid);
219 if (electronPids.has(currentPid)) return true;
220 const current = processesByPid.get(currentPid);
221 if (!current || current.ppid === currentPid) return false;
222 currentPid = current.ppid;
223 }
224 return false;
225}
226
227function orderProcessTree(
228 processes: ReadonlyArray<ResourceTelemetryProcess>,
229 rootPids: ReadonlyArray<number>,
230): ReadonlyArray<ResourceTelemetryProcess> {
231 const processesByPid = new Map(processes.map((process) => [process.identity.pid, process]));
232 const childrenByParent = new Map<number, ResourceTelemetryProcess[]>();
233 for (const process of processes) {
234 const children = childrenByParent.get(process.ppid) ?? [];
235 children.push(process);
236 childrenByParent.set(process.ppid, children);
237 }
238 for (const children of childrenByParent.values()) {
239 children.sort((left, right) => left.identity.pid - right.identity.pid);
240 }
241
242 const ordered: ResourceTelemetryProcess[] = [];
243 const visited = new Set<number>();
244 const visit = (process: ResourceTelemetryProcess): void => {
245 if (visited.has(process.identity.pid)) return;
246 visited.add(process.identity.pid);
247 ordered.push(process);
248 for (const child of childrenByParent.get(process.identity.pid) ?? []) {
249 visit(child);
250 }
251 };
252
253 for (const rootPid of rootPids) {
254 const root = processesByPid.get(rootPid);
255 if (root) visit(root);
256 }
257 for (const process of processes.toSorted(
258 (left, right) => left.depth - right.depth || left.identity.pid - right.identity.pid,
259 )) {
260 visit(process);
261 }
262 return ordered;
263}
264
265function delta(input: {
266 readonly current: number;
267 readonly previous: number;
268 readonly elapsedMs: number;
269}): number {
270 if (
271 input.elapsedMs <= 0 ||
272 input.elapsedMs > MAX_DELTA_INTERVAL_MS ||
273 input.current < input.previous
274 ) {
275 return 0;
276 }
277 return input.current - input.previous;
278}
279
280function incrementCounters(counters: GroupCounters, update: Partial<GroupCounters>): GroupCounters {
281 return {
282 cpuTimeMs: counters.cpuTimeMs + (update.cpuTimeMs ?? 0),
283 ioReadBytes: counters.ioReadBytes + (update.ioReadBytes ?? 0),
284 ioWriteBytes: counters.ioWriteBytes + (update.ioWriteBytes ?? 0),
285 processStarts: counters.processStarts + (update.processStarts ?? 0),
286 processExits: counters.processExits + (update.processExits ?? 0),
287 };
288}
289
290function applyLifecycleCounters(input: {
291 readonly counters: TelemetryCounters;
292 readonly deltas: ReadonlyArray<ProcessDelta>;
293 readonly current: ReadonlyMap<string, ProcessState>;
294 readonly previous: ReadonlyMap<string, ProcessState>;
295}): TelemetryCounters {
296 let backend = input.counters.backend;
297 let electron = input.counters.electron;
298 let monitor = input.counters.monitor;
299 let allT3 = input.counters.allT3;
300 for (const processDelta of input.deltas) {
301 const group = categoryGroup(processDelta.category);
302 switch (group) {
303 case "backend":
304 backend = incrementCounters(backend, processDelta);
305 break;
306 case "electron":
307 electron = incrementCounters(electron, processDelta);
308 break;
309 case "monitor":
310 monitor = incrementCounters(monitor, processDelta);
311 break;
312 }
313 allT3 = incrementCounters(allT3, processDelta);
314 }
315
316 for (const [identityKey, current] of input.current) {
317 if (input.previous.has(identityKey)) continue;
318 const group = categoryGroup(current.process.category);
319 switch (group) {
320 case "backend":
321 backend = incrementCounters(backend, { processStarts: 1 });
322 break;
323 case "electron":
324 electron = incrementCounters(electron, { processStarts: 1 });
325 break;
326 case "monitor":
327 monitor = incrementCounters(monitor, { processStarts: 1 });
328 break;
329 }
330 allT3 = incrementCounters(allT3, { processStarts: 1 });
331 }
332
333 for (const [identityKey, previous] of input.previous) {
334 if (input.current.has(identityKey)) continue;
335 const group = categoryGroup(previous.process.category);
336 switch (group) {
337 case "backend":
338 backend = incrementCounters(backend, { processExits: 1 });
339 break;
340 case "electron":
341 electron = incrementCounters(electron, { processExits: 1 });
342 break;
343 case "monitor":
344 monitor = incrementCounters(monitor, { processExits: 1 });
345 break;
346 }
347 allT3 = incrementCounters(allT3, { processExits: 1 });
348 }
349
350 return { backend, electron, monitor, allT3 };
351}
352
353function aggregate(
354 processes: ReadonlyArray<ResourceTelemetryProcess>,
355 counters: GroupCounters,
356): ResourceTelemetryAggregate {
357 return {
358 processCount: processes.length,
359 currentCpuPercent: processes.reduce((total, process) => total + process.cpuPercent, 0),
360 cpuTimeMs: counters.cpuTimeMs,
361 currentRssBytes: processes.reduce((total, process) => total + process.residentBytes, 0),
362 peakRssBytes: processes.reduce((total, process) => total + process.peakResidentBytes, 0),
363 ioReadBytes: counters.ioReadBytes,
364 ioWriteBytes: counters.ioWriteBytes,
365 ioReadBytesPerSecond: processes.reduce(
366 (total, process) => total + process.ioReadBytesPerSecond,
367 0,
368 ),
369 ioWriteBytesPerSecond: processes.reduce(
370 (total, process) => total + process.ioWriteBytesPerSecond,
371 0,
372 ),
373 processStarts: counters.processStarts,
374 processExits: counters.processExits,
375 };
376}
377
378export function mergeProcesses(input: MergeProcessesInput): MergeProcessesResult {
379 const nativeProcesses = Option.match(input.nativeSnapshot, {
380 onNone: () => [] as ReadonlyArray<ResourceMonitorProcessSample>,
381 onSome: (snapshot) => snapshot.processes,
382 });
383 const electronMetrics = Option.match(input.desktopSnapshot, {
384 onNone: () => [] as ReadonlyArray<DesktopElectronProcessMetric>,
385 onSome: (snapshot) => snapshot.electronProcesses,
386 });
387 const sampledAtMs = Option.match(input.nativeSnapshot, {
388 onNone: () =>
389 Option.match(input.desktopSnapshot, {
390 onNone: () => input.fallbackSampledAtMs,
391 onSome: (snapshot) => snapshot.sampledAtUnixMs,
392 }),
393 onSome: (native) =>
394 Option.match(input.desktopSnapshot, {
395 onNone: () => native.sampledAtUnixMs,
396 onSome: (desktop) => Math.max(native.sampledAtUnixMs, desktop.sampledAtUnixMs),
397 }),
398 });
399 const nativeSampledAtMs = Option.map(
400 input.nativeSnapshot,
401 (snapshot) => snapshot.sampledAtUnixMs,
402 );
403 const desktopSampledAtMs = Option.map(
404 input.desktopSnapshot,
405 (snapshot) => snapshot.sampledAtUnixMs,
406 );
407 const nativeProcessPids = new Set(nativeProcesses.map((process) => process.pid));
408 const nativeByPid = new Map(nativeProcesses.map((process) => [process.pid, process]));
409 const metricsByPid = new Map<number, DesktopElectronProcessMetric>();
410 for (const metric of electronMetrics) {
411 const nativeProcess = nativeByPid.get(metric.pid);
412 if (!nativeProcess) {
413 nativeByPid.set(
414 metric.pid,
415 syntheticNativeSample(
416 metric,
417 Option.getOrElse(desktopSampledAtMs, () => sampledAtMs),
418 input.previous.get(processIdentityKey(metric.pid, metric.creationTimeMs)),
419 ),
420 );
421 metricsByPid.set(metric.pid, metric);
422 continue;
423 }
424 if (
425 Math.abs(metric.creationTimeMs - nativeProcess.startTimeMs) <= ELECTRON_IDENTITY_TOLERANCE_MS
426 ) {
427 metricsByPid.set(metric.pid, metric);
428 }
429 }
430 const processes = [...nativeByPid.values()];
431 const processesByPid = new Map(processes.map((process) => [process.pid, process]));
432 const requestedElectronRootPids = input.electronRootPids ?? new Set<number>();
433 const explicitElectronRootPids = new Set(
434 [...requestedElectronRootPids].filter((pid) => {
435 const process = processesByPid.get(pid);
436 if (!process) return false;
437 const expectedStartTime = input.electronRootStartTimes?.get(pid);
438 if (expectedStartTime !== undefined) {
439 return Math.abs(process.startTimeMs - expectedStartTime) <= ELECTRON_IDENTITY_TOLERANCE_MS;
440 }
441 if (metricsByPid.has(pid)) return true;
442 return [...input.previous.values()].some(
443 (previous) =>
444 previous.process.category === "electron-main" &&
445 previous.process.identity.pid === pid &&
446 previous.process.identity.startTimeMs === process.startTimeMs,
447 );
448 }),
449 );
450 const electronPids = new Set([...metricsByPid.keys(), ...explicitElectronRootPids]);
451 const electronRootPids = [
452 ...explicitElectronRootPids,
453 ...[...electronPids]
454 .filter((pid) => {
455 if (explicitElectronRootPids.has(pid)) return false;
456 const process = processesByPid.get(pid);
457 return process === undefined
458 ? true
459 : !hasElectronAncestor(process, processesByPid, electronPids);
460 })
461 .toSorted((left, right) => left - right),
462 ].filter((pid, index, values) => values.indexOf(pid) === index);
463 const rootPids = [input.serverPid, ...electronRootPids];
464 const roots = new Set(rootPids);
465 const depths = processDepths(processes, roots);
466 const childrenByParent = new Map<number, number[]>();
467 for (const process of processes) {
468 const children = childrenByParent.get(process.ppid) ?? [];
469 children.push(process.pid);
470 childrenByParent.set(process.ppid, children);
471 }
472
473 const nextPrevious = new Map<string, ProcessState>();
474 const processDeltas: ProcessDelta[] = [];
475 const normalized = processes.map((process): ResourceTelemetryProcess => {
476 const identityKey = processIdentityKey(process.pid, process.startTimeMs);
477 const previous = input.previous.get(identityKey);
478 const counterSampledAtMs = nativeProcessPids.has(process.pid)
479 ? Option.getOrElse(nativeSampledAtMs, () => sampledAtMs)
480 : Option.getOrElse(desktopSampledAtMs, () => sampledAtMs);
481 const elapsedMs = previous ? counterSampledAtMs - previous.sampledAtMs : 0;
482 const cpuTimeDelta = previous
483 ? delta({
484 current: process.cpuTimeMs,
485 previous: previous.process.cpuTimeMs,
486 elapsedMs,
487 })
488 : 0;
489 const ioReadDelta = previous
490 ? delta({
491 current: process.ioReadBytes,
492 previous: previous.process.ioReadBytes,
493 elapsedMs,
494 })
495 : 0;
496 const ioWriteDelta = previous
497 ? delta({
498 current: process.ioWriteBytes,
499 previous: previous.process.ioWriteBytes,
500 elapsedMs,
501 })
502 : 0;
503 const electronMetric = matchElectronMetric(process, metricsByPid);
504 const category: ResourceTelemetryProcessCategory =
505 process.pid === input.serverPid
506 ? "server"
507 : Option.contains(input.sidecarPid, process.pid)
508 ? "resource-monitor"
509 : explicitElectronRootPids.has(process.pid)
510 ? "electron-main"
511 : electronMetric
512 ? electronCategory(electronMetric)
513 : isElectronDescendant(process.pid, processesByPid, electronPids, input.serverPid)
514 ? inferredElectronCategory(process)
515 : "server-child";
516 const firstSeenAt = previous?.process.firstSeenAt ?? DateTime.makeUnsafe(sampledAtMs);
517 const preservePreviousRates = !input.updatePrevious && previous !== undefined;
518 const cpuPercent = preservePreviousRates
519 ? previous.process.cpuPercent
520 : previous && elapsedMs > 0 && elapsedMs <= MAX_DELTA_INTERVAL_MS
521 ? (cpuTimeDelta / elapsedMs) * 100
522 : finiteNonNegative(process.cpuPercent);
523 const normalizedProcess: ResourceTelemetryProcess = {
524 identity: {
525 pid: process.pid,
526 startTimeMs: process.startTimeMs,
527 },
528 ppid: process.ppid,
529 childPids: [...(childrenByParent.get(process.pid) ?? [])].toSorted(
530 (left, right) => left - right,
531 ),
532 depth: depths.get(process.pid) ?? 0,
533 name: process.name,
534 command: process.command,
535 status: process.status,
536 category,
537 ...(electronMetric ? { electronType: electronMetric.type } : {}),
538 ...(electronMetric?.serviceName ? { electronServiceName: electronMetric.serviceName } : {}),
539 cpuPercent: finiteNonNegative(cpuPercent),
540 cpuTimeMs: process.cpuTimeMs,
541 residentBytes: process.residentBytes,
542 peakResidentBytes: Math.max(
543 process.residentBytes,
544 electronMetric?.peakWorkingSetBytes ?? 0,
545 previous?.process.peakResidentBytes ?? 0,
546 ),
547 virtualBytes: process.virtualBytes,
548 ioReadBytes: process.ioReadBytes,
549 ioWriteBytes: process.ioWriteBytes,
550 ioReadBytesPerSecond: preservePreviousRates
551 ? previous.process.ioReadBytesPerSecond
552 : elapsedMs > 0
553 ? finiteNonNegative((ioReadDelta * 1_000) / elapsedMs)
554 : 0,
555 ioWriteBytesPerSecond: preservePreviousRates
556 ? previous.process.ioWriteBytesPerSecond
557 : elapsedMs > 0
558 ? finiteNonNegative((ioWriteDelta * 1_000) / elapsedMs)
559 : 0,
560 ioSemantics: process.ioSemantics,
561 ...(electronMetric ? { idleWakeupsPerSecond: electronMetric.idleWakeupsPerSecond } : {}),
562 runTimeMs: process.runTimeMs,
563 firstSeenAt,
564 lastSeenAt: DateTime.makeUnsafe(sampledAtMs),
565 };
566 nextPrevious.set(identityKey, {
567 process: normalizedProcess,
568 sampledAtMs: counterSampledAtMs,
569 });
570 processDeltas.push({
571 identityKey,
572 category,
573 cpuTimeMs: cpuTimeDelta,
574 ioReadBytes: ioReadDelta,
575 ioWriteBytes: ioWriteDelta,
576 });
577 return normalizedProcess;
578 });
579 const ordered = orderProcessTree(normalized, rootPids);
580
581 const counters =
582 input.updatePrevious && input.accumulateCounters !== false
583 ? applyLifecycleCounters({
584 counters: input.counters,
585 deltas: processDeltas,
586 current: nextPrevious,
587 previous: input.previous,
588 })
589 : input.counters;
590 const backendProcesses = ordered.filter(
591 (process) => categoryGroup(process.category) === "backend",
592 );
593 const electronProcesses = ordered.filter(
594 (process) => categoryGroup(process.category) === "electron",
595 );
596 const monitorProcesses = ordered.filter(
597 (process) => categoryGroup(process.category) === "monitor",
598 );
599
600 return {
601 sampledAtMs,
602 processes: ordered,
603 previous: input.updatePrevious ? nextPrevious : input.previous,
604 counters,
605 groups: {
606 backend: aggregate(backendProcesses, counters.backend),
607 electron: aggregate(electronProcesses, counters.electron),
608 monitor: aggregate(monitorProcesses, counters.monitor),
609 allT3: aggregate(ordered, counters.allT3),
610 },
611 deltas: processDeltas,
612 };
613}