Open History to see the last commit
1import type {
2 DesktopHostTelemetrySnapshot,
3 ResourceMonitorProcessSample,
4 ResourceMonitorSnapshotEvent,
5} from "@t3tools/contracts";
6import { describe, expect, it } from "@effect/vitest";
7import * as DateTime from "effect/DateTime";
8import * as Deferred from "effect/Deferred";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Layer from "effect/Layer";
12import * as Option from "effect/Option";
13import * as PubSub from "effect/PubSub";
14import * as Ref from "effect/Ref";
15import * as Stream from "effect/Stream";
16
17import * as DesktopTelemetryReceiver from "./DesktopTelemetryReceiver.ts";
18import * as NativeTelemetryClient from "./NativeTelemetryClient.ts";
19import * as ResourceAttribution from "./ResourceAttribution.ts";
20import * as ResourceTelemetry from "./ResourceTelemetry.ts";
21
22// ResourceTelemetry classifies process.pid as the server, so fixture PIDs are
23// offsets from it. Fixed PIDs collide whenever the test worker is assigned one.
24const SERVER_PID = process.pid;
25const CODEX_PID = SERVER_PID + 1;
26const SHORT_LIVED_CHILD_PID = SERVER_PID + 2;
27const ELECTRON_PID = SERVER_PID + 3;
28const MONITOR_PID = SERVER_PID + 4;
29
30function processSample(
31 input: Partial<ResourceMonitorProcessSample> &
32 Pick<ResourceMonitorProcessSample, "pid" | "ppid" | "startTimeMs">,
33): ResourceMonitorProcessSample {
34 return {
35 runTimeMs: 1_000,
36 name: `process-${input.pid}`,
37 command: `process-${input.pid}`,
38 status: "Running",
39 cpuPercent: 0,
40 cpuTimeMs: 0,
41 residentBytes: 1_024,
42 virtualBytes: 2_048,
43 ioReadBytes: 0,
44 ioWriteBytes: 0,
45 ioSemantics: "storage",
46 ...input,
47 };
48}
49
50function nativeSnapshot(input: {
51 readonly sequence: number;
52 readonly sampledAtUnixMs: number;
53 readonly childCpuTimeMs: number;
54 readonly childWriteBytes: number;
55 readonly externalProcesses?: ResourceMonitorSnapshotEvent["externalProcesses"];
56}): ResourceMonitorSnapshotEvent {
57 const processes = [
58 processSample({
59 pid: SERVER_PID,
60 ppid: 1,
61 startTimeMs: 100,
62 cpuTimeMs: input.sequence * 10,
63 }),
64 processSample({
65 pid: CODEX_PID,
66 ppid: SERVER_PID,
67 startTimeMs: 200,
68 name: "codex",
69 command: "codex app-server",
70 cpuTimeMs: input.childCpuTimeMs,
71 ioWriteBytes: input.childWriteBytes,
72 }),
73 processSample({
74 pid: ELECTRON_PID,
75 ppid: 1,
76 startTimeMs: 300,
77 name: "electron",
78 command: "electron",
79 cpuTimeMs: input.sequence * 20,
80 }),
81 processSample({
82 pid: MONITOR_PID,
83 ppid: SERVER_PID,
84 startTimeMs: 400,
85 name: "t3-resource-monitor",
86 command: "t3-resource-monitor",
87 cpuTimeMs: input.sequence * 5,
88 }),
89 ];
90 return {
91 version: 3,
92 type: "snapshot",
93 sequence: input.sequence,
94 sampledAtUnixMs: input.sampledAtUnixMs,
95 collectionDurationMicros: 300,
96 scannedProcessCount: 80,
97 retainedProcessCount: processes.length,
98 inaccessibleProcessCount: 1,
99 ...(input.externalProcesses === undefined
100 ? {}
101 : { externalProcesses: input.externalProcesses }),
102 processes,
103 };
104}
105
106function nativeGeneration(
107 snapshot: ResourceMonitorSnapshotEvent,
108 generation: number,
109): NativeTelemetryClient.NativeTelemetrySnapshot {
110 return { generation, snapshot };
111}
112
113function desktopSnapshot(sampledAtUnixMs: number): DesktopHostTelemetrySnapshot {
114 const sampledAt = DateTime.makeUnsafe(sampledAtUnixMs);
115 return {
116 version: 1,
117 type: "desktopTelemetry",
118 sequence: 1,
119 sampledAtUnixMs,
120 electronPid: ELECTRON_PID,
121 power: {
122 source: "electron-main",
123 idle: "false",
124 idleSeconds: 2,
125 locked: "false",
126 suspended: false,
127 onBattery: "true",
128 lowPowerMode: "unknown",
129 thermalState: "fair",
130 stale: false,
131 updatedAt: sampledAt,
132 },
133 speedLimitPercent: Option.some(90),
134 electronProcesses: [
135 {
136 pid: ELECTRON_PID,
137 creationTimeMs: 300,
138 type: "Browser",
139 name: "electron",
140 cpuPercent: 2,
141 cumulativeCpuSeconds: 0.02,
142 idleWakeupsPerSecond: 3,
143 workingSetBytes: 4_096,
144 peakWorkingSetBytes: 8_192,
145 },
146 ],
147 };
148}
149
150describe("ResourceTelemetry", () => {
151 it.effect("replays churning history and retains baselines for returning processes", () =>
152 Effect.gen(function* () {
153 const readAtMs = DateTime.toEpochMillis(yield* DateTime.now);
154 const sampleCount = 100;
155 const samples = Array.from({ length: sampleCount }, (_, index) => {
156 const sample = nativeSnapshot({
157 sequence: index + 1,
158 sampledAtUnixMs: readAtMs - (sampleCount - index) * 1_000,
159 childCpuTimeMs: index * 10,
160 childWriteBytes: index * 100,
161 });
162 const processes = sample.processes
163 .filter((entry) => entry.pid === SERVER_PID || entry.pid === CODEX_PID)
164 .map((entry) =>
165 entry.pid === CODEX_PID
166 ? { ...entry, startTimeMs: index }
167 : { ...entry, ioWriteBytes: index * 100 },
168 );
169 if (index === 0 || index === 2) {
170 processes.push(
171 processSample({
172 pid: SHORT_LIVED_CHILD_PID,
173 ppid: SERVER_PID,
174 startTimeMs: 500,
175 cpuTimeMs: index === 0 ? 100 : 350,
176 ioWriteBytes: index === 0 ? 1_000 : 5_000,
177 }),
178 );
179 }
180 return { ...sample, processes };
181 });
182 const layerTelemetry = ResourceTelemetry.layer.pipe(
183 Layer.provide(
184 Layer.mergeAll(
185 NativeTelemetryClient.layerTest({ readHistory: () => Effect.succeed(samples) }),
186 DesktopTelemetryReceiver.layerTest(),
187 ResourceAttribution.layer,
188 ),
189 ),
190 );
191
192 yield* Effect.gen(function* () {
193 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
194 const history = yield* telemetry.readHistory({
195 windowMs: sampleCount * 1_000,
196 bucketMs: 10_000,
197 });
198
199 expect(history.topProcesses).toHaveLength(sampleCount + 2);
200 expect(
201 history.topProcesses.find((entry) => entry.identity.pid === SERVER_PID),
202 ).toMatchObject({
203 sampleCount,
204 cpuTimeMs: (sampleCount - 1) * 10,
205 ioWriteBytes: (sampleCount - 1) * 100,
206 });
207 const children = history.topProcesses.filter((entry) => entry.identity.pid === CODEX_PID);
208 expect(children).toHaveLength(sampleCount);
209 for (const child of children) {
210 expect(child).toMatchObject({ sampleCount: 1, cpuTimeMs: 0, ioWriteBytes: 0 });
211 }
212 expect(
213 history.topProcesses.find((entry) => entry.identity.pid === SHORT_LIVED_CHILD_PID),
214 ).toMatchObject({
215 sampleCount: 2,
216 cpuTimeMs: 250,
217 ioWriteBytes: 4_000,
218 });
219 expect(Math.max(...history.buckets.map((bucket) => bucket.maxProcessCount))).toBe(3);
220 expect(history.buckets.reduce((total, bucket) => total + bucket.ioWriteBytes, 0)).toBe(
221 (sampleCount - 1) * 100 + 4_000,
222 );
223 }).pipe(Effect.provide(layerTelemetry));
224 }),
225 );
226
227 it.effect(
228 "attributes an initial Electron root from the identity recorded by the native snapshot",
229 () =>
230 Effect.gen(function* () {
231 const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now);
232 const sample = nativeSnapshot({
233 sequence: 1,
234 sampledAtUnixMs,
235 childCpuTimeMs: 100,
236 childWriteBytes: 1_000,
237 externalProcesses: [{ pid: ELECTRON_PID, startTimeMs: 300 }],
238 });
239 const desktop = {
240 ...desktopSnapshot(sampledAtUnixMs),
241 electronProcesses: [],
242 };
243 const layerNative = NativeTelemetryClient.layerTest({
244 sampleNow: Effect.succeed(nativeGeneration(sample, 0)),
245 health: Effect.succeed({
246 status: "healthy",
247 hello: Option.none(),
248 lastSampleAt: Option.none(),
249 lastError: Option.none(),
250 restartCount: 0,
251 sampleIntervalMs: 1_000,
252 }),
253 });
254 const layerDesktop = DesktopTelemetryReceiver.layerTest({
255 latest: Effect.succeedSome(desktop),
256 });
257 const layerTelemetry = ResourceTelemetry.layer.pipe(
258 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, ResourceAttribution.layer)),
259 );
260
261 const snapshot = yield* Effect.gen(function* () {
262 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
263 return yield* telemetry.refresh;
264 }).pipe(Effect.provide(layerTelemetry));
265
266 expect(
267 snapshot.processes.find((entry) => entry.identity.pid === ELECTRON_PID)?.category,
268 ).toBe("electron-main");
269 expect(snapshot.groups.electron.processStarts).toBe(1);
270 expect(snapshot.groups.backend.processStarts).toBe(3);
271 }),
272 );
273
274 it.effect("rejects buffered snapshots from an earlier sidecar generation", () =>
275 Effect.scoped(
276 Effect.gen(function* () {
277 const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
278 const stale = nativeSnapshot({
279 sequence: 100,
280 sampledAtUnixMs: startedAt + 1_000,
281 childCpuTimeMs: 100,
282 childWriteBytes: 1_000,
283 });
284 const current = nativeSnapshot({
285 sequence: 1,
286 sampledAtUnixMs: startedAt + 2_000,
287 childCpuTimeMs: 200,
288 childWriteBytes: 2_000,
289 });
290 const nativeSnapshots =
291 yield* PubSub.unbounded<NativeTelemetryClient.NativeTelemetrySnapshot>();
292 const layerNative = NativeTelemetryClient.layerTest({
293 snapshots: Stream.fromPubSub(nativeSnapshots),
294 sampleNow: Effect.never,
295 health: Effect.succeed({
296 status: "healthy",
297 hello: Option.none(),
298 lastSampleAt: Option.none(),
299 lastError: Option.none(),
300 restartCount: 1,
301 sampleIntervalMs: 1_000,
302 }),
303 });
304 const layerTelemetry = ResourceTelemetry.layer.pipe(
305 Layer.provide(
306 Layer.mergeAll(
307 layerNative,
308 DesktopTelemetryReceiver.layerTest(),
309 ResourceAttribution.layer,
310 ),
311 ),
312 );
313
314 yield* Effect.gen(function* () {
315 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
316 const subscription = yield* telemetry.subscribe;
317 const nextSnapshot = yield* Stream.runHead(subscription.changes).pipe(Effect.forkChild);
318
319 yield* PubSub.publish(nativeSnapshots, nativeGeneration(stale, 0));
320 yield* Effect.yieldNow;
321 expect((yield* telemetry.latest).health.scannedProcessCount).toBe(0);
322
323 yield* PubSub.publish(nativeSnapshots, nativeGeneration(current, 1));
324 const received = yield* Fiber.join(nextSnapshot);
325
326 expect(Option.isSome(received)).toBe(true);
327 expect(DateTime.toEpochMillis(Option.getOrThrow(received).readAt)).toBe(
328 current.sampledAtUnixMs,
329 );
330 }).pipe(Effect.provide(layerTelemetry));
331 }),
332 ),
333 );
334
335 it.effect("atomically subscribes while a desktop update is being rebuilt", () =>
336 Effect.scoped(
337 Effect.gen(function* () {
338 const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
339 const nextDesktopSnapshot = desktopSnapshot(startedAt + 1_000);
340 const desktopChanges = yield* PubSub.unbounded<DesktopHostTelemetrySnapshot>();
341 const rebuildStarted = yield* Deferred.make<void>();
342 const finishRebuild = yield* Deferred.make<void>();
343 const attributionReads = yield* Ref.make(0);
344 const layerAttribution = Layer.succeed(
345 ResourceAttribution.ResourceAttribution,
346 ResourceAttribution.ResourceAttribution.of({
347 record: () => Effect.void,
348 snapshot: Ref.updateAndGet(attributionReads, (count) => count + 1).pipe(
349 Effect.flatMap((count) =>
350 count === 1
351 ? Effect.void
352 : Deferred.succeed(rebuildStarted, undefined).pipe(
353 Effect.andThen(Deferred.await(finishRebuild)),
354 ),
355 ),
356 Effect.andThen(
357 Effect.succeed({
358 readAt: DateTime.makeUnsafe(startedAt),
359 entries: [],
360 }),
361 ),
362 ),
363 }),
364 );
365 const layerNative = NativeTelemetryClient.layerTest({
366 sampleNow: Effect.never,
367 health: Effect.succeed({
368 status: "healthy",
369 hello: Option.none(),
370 lastSampleAt: Option.none(),
371 lastError: Option.none(),
372 restartCount: 0,
373 sampleIntervalMs: 1_000,
374 }),
375 });
376 const layerDesktop = DesktopTelemetryReceiver.layerTest({
377 changes: Stream.fromPubSub(desktopChanges),
378 });
379 const layerTelemetry = ResourceTelemetry.layer.pipe(
380 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, layerAttribution)),
381 );
382
383 yield* Effect.gen(function* () {
384 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
385 yield* Effect.yieldNow;
386 yield* PubSub.publish(desktopChanges, nextDesktopSnapshot);
387 yield* Deferred.await(rebuildStarted);
388
389 const subscriptionFiber = yield* telemetry.subscribe.pipe(Effect.forkChild);
390 yield* Effect.yieldNow;
391 yield* Deferred.succeed(finishRebuild, undefined);
392 const subscription = yield* Fiber.join(subscriptionFiber);
393
394 expect(DateTime.toEpochMillis(subscription.latest.readAt)).toBe(
395 nextDesktopSnapshot.sampledAtUnixMs,
396 );
397 expect(subscription.latest.power).toEqual(nextDesktopSnapshot.power);
398 }).pipe(Effect.provide(layerTelemetry));
399 }),
400 ),
401 );
402});