Open History to see the last commit
1import type {
2 HostPowerSnapshot,
3 ResourceMonitorCapabilities,
4 ResourceMonitorCommand,
5 ResourceMonitorEvent,
6 ResourceMonitorExternalProcess,
7 ResourceMonitorHelloEvent,
8 ResourceMonitorProcessTableEntry,
9 ResourceMonitorSnapshotEvent,
10 ResourceTelemetrySourceStatus,
11} from "@t3tools/contracts";
12import {
13 RESOURCE_MONITOR_PROTOCOL_VERSION,
14 ResourceMonitorCommand as ResourceMonitorCommandSchema,
15 ResourceMonitorEvent as ResourceMonitorEventSchema,
16} from "@t3tools/contracts";
17import * as Cause from "effect/Cause";
18import * as Context from "effect/Context";
19import * as SecureRandom from "effect/crypto/SecureRandom";
20import * as DateTime from "effect/DateTime";
21import * as Deferred from "effect/Deferred";
22import * as Duration from "effect/Duration";
23import * as Effect from "effect/Effect";
24import * as Fiber from "effect/Fiber";
25import * as Layer from "effect/Layer";
26import * as Option from "effect/Option";
27import * as PubSub from "effect/PubSub";
28import * as Queue from "effect/Queue";
29import * as Ref from "effect/Ref";
30import * as Result from "effect/Result";
31import * as Schema from "effect/Schema";
32import * as Semaphore from "effect/Semaphore";
33import * as Scope from "effect/Scope";
34import * as Stream from "effect/Stream";
35import * as Ndjson from "effect/encoding/Ndjson";
36import { ChildProcess, ChildProcessSpawner } from "effect/process";
37
38import * as ResourceMonitorBinary from "./ResourceMonitorBinary.ts";
39import * as ServerConfig from "../config.ts";
40import { subscribeBeforeSnapshotWithoutMutex } from "../utils/subscribeBeforeSnapshot.ts";
41
42const SAMPLE_INTERVAL_MS = 1_000;
43// The sidecar treats 0 as "sample only on request".
44const IDLE_SAMPLE_INTERVAL_MS = 0;
45const BATTERY_SAMPLE_INTERVAL_MS = 5_000;
46const CONSTRAINED_SAMPLE_INTERVAL_MS = 15_000;
47const HANDSHAKE_TIMEOUT = Duration.seconds(5);
48const SAMPLE_REQUEST_TIMEOUT = Duration.seconds(5);
49const PROCESS_TABLE_REQUEST_TIMEOUT = Duration.seconds(5);
50const HISTORY_REQUEST_TIMEOUT = Duration.seconds(15);
51const INITIAL_RESTART_DELAY = Duration.millis(500);
52const MAX_RESTART_DELAY = Duration.seconds(10);
53const FAILURE_WINDOW_MS = 60_000;
54const MAX_FAILURES_PER_WINDOW = 5;
55
56export class NativeTelemetrySpawnFailed extends Schema.TaggedError<NativeTelemetrySpawnFailed>()(
57 "NativeTelemetrySpawnFailed",
58 {
59 path: Schema.String,
60 cause: Schema.Defect(),
61 },
62) {
63 override get message(): string {
64 return `Failed to start resource monitor '${this.path}'.`;
65 }
66}
67
68export class NativeTelemetryHandshakeTimedOut extends Schema.TaggedError<NativeTelemetryHandshakeTimedOut>()(
69 "NativeTelemetryHandshakeTimedOut",
70 {
71 timeoutMs: Schema.Number,
72 },
73) {
74 override get message(): string {
75 return `Resource monitor handshake timed out after ${this.timeoutMs}ms.`;
76 }
77}
78
79class NativeTelemetryRequestTimedOut extends Schema.TaggedError<NativeTelemetryRequestTimedOut>()(
80 "NativeTelemetryRequestTimedOut",
81 {
82 operation: Schema.Literals(["processTable", "readHistory", "sampleNow"]),
83 timeoutMs: Schema.Number,
84 },
85) {
86 override get message(): string {
87 return `Resource monitor '${this.operation}' request timed out after ${this.timeoutMs}ms.`;
88 }
89}
90
91export class NativeTelemetryProtocolMismatch extends Schema.TaggedError<NativeTelemetryProtocolMismatch>()(
92 "NativeTelemetryProtocolMismatch",
93 {
94 expectedVersion: Schema.Number,
95 receivedVersion: Schema.Number,
96 },
97) {
98 override get message(): string {
99 return `Resource monitor protocol ${this.receivedVersion} is incompatible with expected protocol ${this.expectedVersion}.`;
100 }
101}
102
103export class NativeTelemetryDecodeFailed extends Schema.TaggedError<NativeTelemetryDecodeFailed>()(
104 "NativeTelemetryDecodeFailed",
105 {
106 cause: Schema.Defect(),
107 },
108) {
109 override get message(): string {
110 return "Failed to decode resource monitor output.";
111 }
112}
113
114export class NativeTelemetryCommandFailed extends Schema.TaggedError<NativeTelemetryCommandFailed>()(
115 "NativeTelemetryCommandFailed",
116 {
117 operation: Schema.String,
118 cause: Schema.Defect(),
119 },
120) {
121 override get message(): string {
122 return `Resource monitor command '${this.operation}' failed.`;
123 }
124}
125
126export class NativeTelemetryExited extends Schema.TaggedError<NativeTelemetryExited>()(
127 "NativeTelemetryExited",
128 {
129 exitCode: Schema.Number,
130 },
131) {
132 override get message(): string {
133 return `Resource monitor exited with code ${this.exitCode}.`;
134 }
135}
136
137class NativeTelemetryStreamClosed extends Schema.TaggedError<NativeTelemetryStreamClosed>()(
138 "NativeTelemetryStreamClosed",
139 {},
140) {
141 override get message(): string {
142 return "Resource monitor event stream closed unexpectedly.";
143 }
144}
145
146export class NativeTelemetryUnavailable extends Schema.TaggedError<NativeTelemetryUnavailable>()(
147 "NativeTelemetryUnavailable",
148 {
149 reason: Schema.String,
150 },
151) {
152 override get message(): string {
153 return `Resource monitor is unavailable: ${this.reason}`;
154 }
155}
156
157export type NativeTelemetryClientError =
158 | ResourceMonitorBinary.ResourceMonitorBinaryError
159 | NativeTelemetrySpawnFailed
160 | NativeTelemetryHandshakeTimedOut
161 | NativeTelemetryRequestTimedOut
162 | NativeTelemetryProtocolMismatch
163 | NativeTelemetryDecodeFailed
164 | NativeTelemetryCommandFailed
165 | NativeTelemetryExited
166 | NativeTelemetryStreamClosed
167 | NativeTelemetryUnavailable;
168
169export interface NativeTelemetryClientHealth {
170 readonly status: ResourceTelemetrySourceStatus;
171 readonly hello: Option.Option<ResourceMonitorHelloEvent>;
172 readonly lastSampleAt: Option.Option<DateTime.Utc>;
173 readonly lastError: Option.Option<string>;
174 readonly restartCount: number;
175 readonly sampleIntervalMs: number;
176}
177
178export interface NativeTelemetrySnapshot {
179 readonly generation: number;
180 readonly snapshot: ResourceMonitorSnapshotEvent;
181}
182
183export class NativeTelemetryClient extends Context.Service<
184 NativeTelemetryClient,
185 {
186 readonly capabilities: Effect.Effect<ResourceMonitorCapabilities, NativeTelemetryClientError>;
187 readonly snapshots: Stream.Stream<NativeTelemetrySnapshot, NativeTelemetryClientError>;
188 readonly readHistory: (
189 windowMs: number,
190 ) => Effect.Effect<ReadonlyArray<ResourceMonitorSnapshotEvent>, NativeTelemetryClientError>;
191 readonly setExternalProcesses: (
192 processes: ReadonlyArray<ResourceMonitorExternalProcess>,
193 ) => Effect.Effect<void, NativeTelemetryClientError>;
194 readonly setHostPowerState: (
195 snapshot: HostPowerSnapshot,
196 ) => Effect.Effect<void, NativeTelemetryClientError>;
197 readonly sampleNow: Effect.Effect<NativeTelemetrySnapshot, NativeTelemetryClientError>;
198 readonly processTable: Effect.Effect<
199 ReadonlyArray<ResourceMonitorProcessTableEntry>,
200 NativeTelemetryClientError
201 >;
202 readonly retry: Effect.Effect<boolean>;
203 readonly health: Effect.Effect<NativeTelemetryClientHealth>;
204 readonly subscribeHealth: Effect.Effect<
205 {
206 readonly latest: NativeTelemetryClientHealth;
207 readonly changes: Stream.Stream<NativeTelemetryClientHealth>;
208 },
209 never,
210 Scope.Scope
211 >;
212 }
213>()("t3/resourceTelemetry/NativeTelemetryClient") {}
214
215interface ClientState {
216 readonly status: ResourceTelemetrySourceStatus;
217 readonly handle: Option.Option<ChildProcessSpawner.ChildProcessHandle>;
218 readonly hello: Option.Option<ResourceMonitorHelloEvent>;
219 readonly lastSampleAt: Option.Option<DateTime.Utc>;
220 readonly lastError: Option.Option<string>;
221 readonly restartCount: number;
222}
223
224export interface CollectionControl {
225 readonly hostPower: HostPowerSnapshot;
226 readonly liveSubscriberCount: number;
227 readonly sampleIntervalMs: number;
228}
229
230interface PendingHistoryRequest {
231 readonly deferred: Deferred.Deferred<
232 ReadonlyArray<ResourceMonitorSnapshotEvent>,
233 NativeTelemetryClientError
234 >;
235 readonly snapshots: ReadonlyArray<ResourceMonitorSnapshotEvent>;
236}
237
238const initialState: ClientState = {
239 status: "starting",
240 handle: Option.none(),
241 hello: Option.none(),
242 lastSampleAt: Option.none(),
243 lastError: Option.none(),
244 restartCount: 0,
245};
246
247function toHealth(state: ClientState, sampleIntervalMs: number): NativeTelemetryClientHealth {
248 return {
249 status: state.status,
250 hello: state.hello,
251 lastSampleAt: state.lastSampleAt,
252 lastError: state.lastError,
253 restartCount: state.restartCount,
254 sampleIntervalMs,
255 };
256}
257
258function isThermallyConstrained(snapshot: HostPowerSnapshot): boolean {
259 return snapshot.thermalState === "serious" || snapshot.thermalState === "critical";
260}
261
262// Without a live subscriber the sidecar does not sample on a timer, so an idle server
263// never walks the process table. History then covers only time with a subscriber.
264function resolveNativeSampleIntervalMs(
265 snapshot: HostPowerSnapshot,
266 liveSubscriberCount: number,
267): number {
268 if (liveSubscriberCount === 0) return IDLE_SAMPLE_INTERVAL_MS;
269 if (snapshot.stale || snapshot.source === "unknown") return SAMPLE_INTERVAL_MS;
270 if (
271 snapshot.suspended ||
272 snapshot.locked === "true" ||
273 snapshot.lowPowerMode === "true" ||
274 isThermallyConstrained(snapshot)
275 ) {
276 return CONSTRAINED_SAMPLE_INTERVAL_MS;
277 }
278 if (snapshot.onBattery === "true") return BATTERY_SAMPLE_INTERVAL_MS;
279 return SAMPLE_INTERVAL_MS;
280}
281
282// Desktop power samples also commit control state; avoid a span for each sample.
283export const commitCollectionControlUpdate = Effect.fnUntraced(function* <E, R>(
284 desiredState: Ref.Ref<CollectionControl>,
285 appliedState: Ref.Ref<CollectionControl>,
286 update: (current: CollectionControl) => CollectionControl,
287 apply: (previous: CollectionControl, next: CollectionControl) => Effect.Effect<void, E, R>,
288): Effect.fn.Return<readonly [CollectionControl, CollectionControl], E, R> {
289 const [previousDesired, next] = yield* Ref.modify(desiredState, (previous) => {
290 const next = update(previous);
291 return [[previous, next] as const, next];
292 });
293 const previousApplied = yield* Ref.get(appliedState);
294 yield* apply(previousApplied, next);
295 yield* Ref.set(appliedState, next);
296 return [previousDesired, next] as const;
297});
298
299export function synchronizeCollectionControlOnStart<E1, R1, E2, R2>(
300 mutex: Semaphore.Semaphore,
301 desiredState: Ref.Ref<CollectionControl>,
302 appliedState: Ref.Ref<CollectionControl>,
303 apply: (control: CollectionControl) => Effect.Effect<void, E1, R1>,
304 markReady: Effect.Effect<void, E2, R2>,
305) {
306 return mutex.withPermits(1)(
307 Effect.gen(function* () {
308 const control = yield* Ref.get(desiredState);
309 yield* apply(control);
310 yield* Ref.set(appliedState, control);
311 yield* markReady;
312 return control;
313 }),
314 );
315}
316
317const decodeMonitorEvent: (
318 value: unknown,
319) => Effect.Effect<ResourceMonitorEvent, Schema.SchemaError> = Schema.decodeUnknownEffect(
320 ResourceMonitorEventSchema,
321);
322const encodeMonitorCommand = Schema.encodeEffect(
323 Schema.fromJsonString(ResourceMonitorCommandSchema),
324);
325const isProtocolMismatch = Schema.is(NativeTelemetryProtocolMismatch);
326const isDecodeFailed = Schema.is(NativeTelemetryDecodeFailed);
327const isCommandFailed = Schema.is(NativeTelemetryCommandFailed);
328
329export function eventVersion(value: unknown): number | undefined {
330 if (typeof value !== "object" || value === null) return undefined;
331 const version = Reflect.get(value, "version");
332 return typeof version === "number" ? version : undefined;
333}
334
335function restartDelay(attempt: number): Duration.Duration {
336 return Duration.min(Duration.times(INITIAL_RESTART_DELAY, 2 ** attempt), MAX_RESTART_DELAY);
337}
338
339function retainRecentNativeTelemetryFailures(
340 failures: ReadonlyArray<number>,
341 now: number,
342): ReadonlyArray<number> {
343 return failures.filter((failedAt) => now - failedAt <= FAILURE_WINDOW_MS);
344}
345
346function errorMessage(error: NativeTelemetryClientError): string {
347 return error.message;
348}
349
350function canRequestNativeTelemetryRetry(
351 status: ResourceTelemetrySourceStatus,
352 hasHandle: boolean,
353): boolean {
354 return status !== "healthy" && status !== "starting" && !hasHandle;
355}
356
357function canCommandNativeTelemetrySidecar(
358 status: ResourceTelemetrySourceStatus,
359 hasHandle: boolean,
360): boolean {
361 return hasHandle && (status === "healthy" || status === "degraded");
362}
363
364/** @public */
365export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(function* () {
366 const binary = yield* ResourceMonitorBinary.ResourceMonitorBinary;
367 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
368 const random = yield* SecureRandom.SecureRandom;
369 const config = yield* ServerConfig.ServerConfig;
370 const initializedAt = yield* DateTime.now;
371 const state = yield* Ref.make(initialState);
372 const collectionControl = yield* Ref.make<CollectionControl>({
373 hostPower: {
374 source: "unknown",
375 idle: "unknown",
376 idleSeconds: null,
377 locked: "unknown",
378 suspended: false,
379 onBattery: "unknown",
380 lowPowerMode: "unknown",
381 thermalState: "unknown",
382 stale: true,
383 updatedAt: initializedAt,
384 },
385 liveSubscriberCount: 0,
386 sampleIntervalMs: IDLE_SAMPLE_INTERVAL_MS,
387 });
388 const appliedCollectionControl = yield* Ref.make(yield* Ref.get(collectionControl));
389 const externalProcesses = yield* Ref.make<ReadonlyArray<ResourceMonitorExternalProcess>>([]);
390 const pendingSamples = yield* Ref.make(
391 new Map<string, Deferred.Deferred<NativeTelemetrySnapshot, NativeTelemetryClientError>>(),
392 );
393 const pendingProcessTables = yield* Ref.make(
394 new Map<
395 string,
396 Deferred.Deferred<ReadonlyArray<ResourceMonitorProcessTableEntry>, NativeTelemetryClientError>
397 >(),
398 );
399 const pendingHistories = yield* Ref.make(new Map<string, PendingHistoryRequest>());
400 const snapshots = yield* PubSub.sliding<NativeTelemetrySnapshot>(8);
401 const healthChanges = yield* PubSub.sliding<NativeTelemetryClientHealth>(4);
402 const retryQueue = yield* Queue.sliding<void>(1);
403 const commandMutex = yield* Semaphore.make(1);
404 const controlMutex = yield* Semaphore.make(1);
405 const currentHealth = Effect.all([Ref.get(state), Ref.get(collectionControl)]).pipe(
406 Effect.map(([current, control]) => toHealth(current, control.sampleIntervalMs)),
407 );
408 const publishHealth = currentHealth.pipe(
409 Effect.flatMap((health) => PubSub.publish(healthChanges, health)),
410 Effect.asVoid,
411 );
412
413 const failPending = Effect.fn("NativeTelemetryClient.failPending")(function* (
414 error: NativeTelemetryClientError,
415 ) {
416 const samples = yield* Ref.getAndSet(pendingSamples, new Map());
417 const processTables = yield* Ref.getAndSet(pendingProcessTables, new Map());
418 const histories = yield* Ref.getAndSet(pendingHistories, new Map());
419 yield* Effect.forEach(samples.values(), (deferred) => Deferred.fail(deferred, error), {
420 discard: true,
421 });
422 yield* Effect.forEach(processTables.values(), (deferred) => Deferred.fail(deferred, error), {
423 discard: true,
424 });
425 yield* Effect.forEach(histories.values(), (request) => Deferred.fail(request.deferred, error), {
426 discard: true,
427 });
428 });
429
430 const writeCommand = (
431 handle: ChildProcessSpawner.ChildProcessHandle,
432 command: ResourceMonitorCommand,
433 ): Effect.Effect<void, NativeTelemetryClientError> =>
434 commandMutex.withPermits(1)(
435 encodeMonitorCommand(command).pipe(
436 Effect.map((encoded) => `${encoded}\n`),
437 Effect.mapError(
438 (cause) =>
439 new NativeTelemetryCommandFailed({
440 operation: command.type,
441 cause,
442 }),
443 ),
444 Effect.flatMap((encoded) =>
445 Stream.run(Stream.encodeText(Stream.make(encoded)), handle.stdin),
446 ),
447 Effect.mapError(
448 (cause) =>
449 new NativeTelemetryCommandFailed({
450 operation: command.type,
451 cause,
452 }),
453 ),
454 ),
455 );
456
457 const processEvent = (
458 event: ResourceMonitorEvent,
459 helloDeferred: Deferred.Deferred<ResourceMonitorHelloEvent>,
460 generation: number,
461 ): Effect.Effect<void, NativeTelemetryClientError> => {
462 switch (event.type) {
463 case "hello":
464 return Ref.update(state, (current) => ({
465 ...current,
466 status: "starting" as const,
467 hello: Option.some(event),
468 lastError: Option.none(),
469 })).pipe(
470 Effect.andThen(publishHealth),
471 Effect.andThen(Deferred.succeed(helloDeferred, event)),
472 Effect.asVoid,
473 );
474 case "snapshot":
475 return Effect.gen(function* () {
476 const nativeSnapshot = { generation, snapshot: event } satisfies NativeTelemetrySnapshot;
477 const sampledAt = DateTime.makeUnsafe(event.sampledAtUnixMs);
478 const healthChanged = yield* Ref.modify(state, (current) => [
479 current.status !== "healthy" || Option.isSome(current.lastError),
480 {
481 ...current,
482 status: "healthy" as const,
483 lastSampleAt: Option.some(sampledAt),
484 lastError: Option.none(),
485 },
486 ]);
487 if (healthChanged) yield* publishHealth;
488 yield* PubSub.publish(snapshots, nativeSnapshot);
489 if (event.requestId) {
490 const deferred = yield* Ref.modify(pendingSamples, (pending) => {
491 const next = new Map(pending);
492 const current = next.get(event.requestId!);
493 next.delete(event.requestId!);
494 return [Option.fromUndefinedOr(current), next];
495 });
496 if (Option.isSome(deferred)) {
497 yield* Deferred.succeed(deferred.value, nativeSnapshot);
498 }
499 }
500 });
501 case "processTable":
502 return Ref.modify(pendingProcessTables, (pending) => {
503 const next = new Map(pending);
504 const deferred = next.get(event.requestId);
505 next.delete(event.requestId);
506 return [Option.fromUndefinedOr(deferred), next] as const;
507 }).pipe(
508 Effect.flatMap(
509 Option.match({
510 onNone: () => Effect.void,
511 onSome: (deferred) => Deferred.succeed(deferred, event.processes),
512 }),
513 ),
514 Effect.asVoid,
515 );
516 case "historyChunk":
517 return Effect.gen(function* () {
518 const latestSnapshot = event.snapshots.at(-1);
519 const healthChanged = yield* Ref.modify(state, (current) => [
520 current.status !== "healthy" || Option.isSome(current.lastError),
521 {
522 ...current,
523 status: "healthy" as const,
524 lastSampleAt: latestSnapshot
525 ? Option.some(DateTime.makeUnsafe(latestSnapshot.sampledAtUnixMs))
526 : current.lastSampleAt,
527 lastError: Option.none(),
528 },
529 ]);
530 if (healthChanged) yield* publishHealth;
531 const completed = yield* Ref.modify(pendingHistories, (pending) => {
532 const request = pending.get(event.requestId);
533 if (!request) return [Option.none(), pending] as const;
534 const snapshots = [...request.snapshots, ...event.snapshots];
535 const next = new Map(pending);
536 if (event.done) {
537 next.delete(event.requestId);
538 return [Option.some({ deferred: request.deferred, snapshots }), next] as const;
539 }
540 next.set(event.requestId, { deferred: request.deferred, snapshots });
541 return [Option.none(), next] as const;
542 });
543 if (Option.isSome(completed)) {
544 yield* Deferred.succeed(completed.value.deferred, completed.value.snapshots);
545 }
546 });
547 case "error":
548 return Ref.update(state, (current) => ({
549 ...current,
550 status: "degraded" as const,
551 lastError: Option.some(event.message),
552 })).pipe(
553 Effect.andThen(publishHealth),
554 Effect.andThen(
555 event.recoverable
556 ? Effect.void
557 : Effect.fail(
558 new NativeTelemetryCommandFailed({
559 operation: event.code,
560 cause: event.message,
561 }),
562 ),
563 ),
564 );
565 }
566 };
567
568 const runAttempt: Effect.Effect<void, NativeTelemetryClientError> = Effect.scoped(
569 Effect.gen(function* () {
570 const executablePath = yield* binary.resolve;
571 const command = ChildProcess.make(executablePath, [], {
572 cwd: config.cwd,
573 stdin: {
574 stream: "pipe",
575 endOnDone: false,
576 },
577 stdout: "pipe",
578 stderr: "pipe",
579 killSignal: "SIGTERM",
580 forceKillAfter: Duration.seconds(2),
581 });
582 const handle = yield* Effect.acquireRelease(
583 spawner
584 .spawn(command)
585 .pipe(
586 Effect.mapError(
587 (cause) => new NativeTelemetrySpawnFailed({ path: executablePath, cause }),
588 ),
589 ),
590 (child) => child.kill().pipe(Effect.ignore),
591 );
592 yield* Ref.update(state, (current) => ({
593 ...current,
594 status: "starting" as const,
595 handle: Option.some(handle),
596 hello: Option.none(),
597 }));
598 yield* publishHealth;
599 const generation = (yield* Ref.get(state)).restartCount;
600
601 const helloDeferred = yield* Deferred.make<ResourceMonitorHelloEvent>();
602 const eventFiber = yield* handle.stdout.pipe(
603 Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),
604 Stream.mapEffect(
605 (
606 value,
607 ): Effect.Effect<
608 ResourceMonitorEvent,
609 NativeTelemetryProtocolMismatch | NativeTelemetryDecodeFailed
610 > => {
611 const version = eventVersion(value);
612 if (version !== undefined && version !== RESOURCE_MONITOR_PROTOCOL_VERSION) {
613 return Effect.fail(
614 new NativeTelemetryProtocolMismatch({
615 expectedVersion: RESOURCE_MONITOR_PROTOCOL_VERSION,
616 receivedVersion: version,
617 }),
618 );
619 }
620 return decodeMonitorEvent(value).pipe(
621 Effect.mapError((cause) => new NativeTelemetryDecodeFailed({ cause })),
622 );
623 },
624 ),
625 Stream.runForEach((event) => processEvent(event, helloDeferred, generation)),
626 Effect.mapError((cause) =>
627 isProtocolMismatch(cause) || isDecodeFailed(cause) || isCommandFailed(cause)
628 ? cause
629 : new NativeTelemetryDecodeFailed({ cause }),
630 ),
631 Effect.forkScoped,
632 );
633 yield* handle.stderr.pipe(Stream.runDrain, Effect.ignore, Effect.forkScoped);
634
635 const hello = yield* Deferred.await(helloDeferred).pipe(
636 Effect.timeoutOption(HANDSHAKE_TIMEOUT),
637 Effect.flatMap(
638 Option.match({
639 onNone: () =>
640 Effect.fail(
641 new NativeTelemetryHandshakeTimedOut({
642 timeoutMs: Duration.toMillis(HANDSHAKE_TIMEOUT),
643 }),
644 ),
645 onSome: Effect.succeed,
646 }),
647 ),
648 );
649 yield* synchronizeCollectionControlOnStart(
650 controlMutex,
651 collectionControl,
652 appliedCollectionControl,
653 (control) =>
654 Effect.gen(function* () {
655 yield* writeCommand(handle, {
656 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
657 type: "configure",
658 rootPid: process.pid,
659 sampleIntervalMs: control.sampleIntervalMs,
660 externalProcesses: [...(yield* Ref.get(externalProcesses))],
661 });
662 if (control.liveSubscriberCount > 0) {
663 yield* writeCommand(handle, {
664 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
665 type: "setStreaming",
666 enabled: true,
667 });
668 }
669 }),
670 Ref.update(state, (current) => ({
671 ...current,
672 status: "healthy" as const,
673 hello: Option.some(hello),
674 })).pipe(Effect.andThen(publishHealth)),
675 );
676
677 yield* writeCommand(handle, {
678 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
679 type: "setExternalProcesses",
680 processes: [...(yield* Ref.get(externalProcesses))],
681 });
682
683 const exitEffect = handle.exitCode.pipe(
684 Effect.mapError(
685 (cause) =>
686 new NativeTelemetryCommandFailed({
687 operation: "waitForExit",
688 cause,
689 }),
690 ),
691 Effect.flatMap((exitCode) =>
692 Effect.fail(new NativeTelemetryExited({ exitCode: Number(exitCode) })),
693 ),
694 );
695 const decoderEffect = Fiber.join(eventFiber).pipe(
696 Effect.andThen(Effect.fail(new NativeTelemetryStreamClosed())),
697 );
698 return yield* Effect.raceFirst(exitEffect, decoderEffect);
699 }),
700 ).pipe(
701 Effect.ensuring(
702 Ref.update(state, (current) => ({
703 ...current,
704 handle: Option.none(),
705 })),
706 ),
707 );
708
709 yield* Effect.gen(function* () {
710 let failures: ReadonlyArray<number> = [];
711 let restartAttempt = 0;
712
713 while (true) {
714 const result = yield* Effect.result(runAttempt);
715 if (Result.isSuccess(result)) {
716 return;
717 }
718
719 const error = result.failure;
720 const now = DateTime.toEpochMillis(yield* DateTime.now);
721 const recentFailures = retainRecentNativeTelemetryFailures(failures, now);
722 if (recentFailures.length === 0) {
723 restartAttempt = 0;
724 }
725 failures = [...recentFailures, now];
726 const exhausted = failures.length >= MAX_FAILURES_PER_WINDOW;
727 yield* Ref.update(state, (current) => ({
728 ...current,
729 status: exhausted ? ("unavailable" as const) : ("degraded" as const),
730 hello: Option.none(),
731 lastError: Option.some(errorMessage(error)),
732 restartCount: current.restartCount + 1,
733 }));
734 yield* publishHealth;
735 yield* failPending(error);
736
737 if (exhausted) {
738 yield* Queue.take(retryQueue);
739 failures = [];
740 restartAttempt = 0;
741 yield* Ref.update(state, (current) => ({
742 ...current,
743 status: "starting" as const,
744 hello: Option.none(),
745 lastError: Option.none(),
746 }));
747 yield* publishHealth;
748 continue;
749 }
750
751 const manuallyRetried = yield* Effect.raceFirst(
752 Effect.sleep(restartDelay(restartAttempt)).pipe(Effect.as(false)),
753 Queue.take(retryQueue).pipe(Effect.as(true)),
754 );
755 restartAttempt = manuallyRetried ? 0 : restartAttempt + 1;
756 }
757 }).pipe(
758 Effect.catchCause((cause) =>
759 Cause.hasInterruptsOnly(cause)
760 ? Effect.void
761 : Ref.update(state, (current) => ({
762 ...current,
763 status: "unavailable" as const,
764 hello: Option.none(),
765 lastError: Option.some("Resource monitor supervisor stopped unexpectedly."),
766 })).pipe(
767 Effect.andThen(publishHealth),
768 Effect.andThen(
769 Effect.logWarning("Resource monitor supervisor failed", {
770 cause: Cause.pretty(cause),
771 }),
772 ),
773 ),
774 ),
775 Effect.forkScoped,
776 );
777
778 // Desktop power samples reach this even when the control state has not changed.
779 const applyCollectionControl = Effect.fnUntraced(function* (
780 previous: CollectionControl,
781 next: CollectionControl,
782 ) {
783 const current = yield* Ref.get(state);
784 if (canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) {
785 const handle = Option.getOrThrow(current.handle);
786 if (previous.sampleIntervalMs !== next.sampleIntervalMs) {
787 yield* writeCommand(handle, {
788 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
789 type: "setSampleInterval",
790 sampleIntervalMs: next.sampleIntervalMs,
791 });
792 }
793 const wasStreaming = previous.liveSubscriberCount > 0;
794 const isStreaming = next.liveSubscriberCount > 0;
795 if (wasStreaming !== isStreaming) {
796 yield* writeCommand(handle, {
797 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
798 type: "setStreaming",
799 enabled: isStreaming,
800 });
801 }
802 }
803 });
804
805 const updateCollectionControl = (update: (current: CollectionControl) => CollectionControl) =>
806 controlMutex.withPermits(1)(
807 commitCollectionControlUpdate(
808 collectionControl,
809 appliedCollectionControl,
810 update,
811 applyCollectionControl,
812 ).pipe(Effect.ensuring(publishHealth), Effect.asVoid),
813 );
814
815 const setHostPowerState: NativeTelemetryClient["Service"]["setHostPowerState"] = (hostPower) =>
816 updateCollectionControl((current) => ({
817 ...current,
818 hostPower,
819 sampleIntervalMs: resolveNativeSampleIntervalMs(hostPower, current.liveSubscriberCount),
820 }));
821
822 const changeLiveSubscriberCount = Effect.fn(
823 "resourceTelemetry.nativeTelemetryClient.changeLiveSubscriberCount",
824 )(function* (delta: 1 | -1) {
825 yield* updateCollectionControl((current) => {
826 const liveSubscriberCount = Math.max(0, current.liveSubscriberCount + delta);
827 return {
828 ...current,
829 liveSubscriberCount,
830 sampleIntervalMs: resolveNativeSampleIntervalMs(current.hostPower, liveSubscriberCount),
831 };
832 });
833 });
834
835 const liveSnapshots = Stream.unwrap(
836 Effect.gen(function* () {
837 const subscription = yield* PubSub.subscribe(snapshots);
838 yield* Effect.acquireRelease(changeLiveSubscriberCount(1), () =>
839 changeLiveSubscriberCount(-1).pipe(Effect.ignore),
840 );
841 return Stream.fromSubscription(subscription);
842 }),
843 );
844
845 // Desktop telemetry refreshes this process set on every sample.
846 const setExternalProcesses: NativeTelemetryClient["Service"]["setExternalProcesses"] =
847 Effect.fnUntraced(function* (processes) {
848 yield* Ref.set(externalProcesses, [...processes]);
849 const current = yield* Ref.get(state);
850 if (!canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) return;
851 yield* writeCommand(Option.getOrThrow(current.handle), {
852 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
853 type: "setExternalProcesses",
854 processes: [...processes],
855 });
856 });
857
858 const readHistory: NativeTelemetryClient["Service"]["readHistory"] = Effect.fn(
859 "NativeTelemetryClient.readHistory",
860 )(function* (windowMs) {
861 const current = yield* Ref.get(state);
862 if (!canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) {
863 return yield* new NativeTelemetryUnavailable({
864 reason: Option.getOrElse(current.lastError, () => "sidecar is not running"),
865 });
866 }
867 const requestId = yield* random.uuidV4().pipe(
868 Effect.mapError(
869 (cause) =>
870 new NativeTelemetryCommandFailed({
871 operation: "createHistoryRequestId",
872 cause,
873 }),
874 ),
875 );
876 const deferred = yield* Deferred.make<
877 ReadonlyArray<ResourceMonitorSnapshotEvent>,
878 NativeTelemetryClientError
879 >();
880 yield* Ref.update(pendingHistories, (pending) => {
881 const next = new Map(pending);
882 next.set(requestId, { deferred, snapshots: [] });
883 return next;
884 });
885 return yield* writeCommand(Option.getOrThrow(current.handle), {
886 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
887 type: "readHistory",
888 requestId,
889 windowMs: Math.max(0, Math.round(windowMs)),
890 }).pipe(
891 Effect.andThen(
892 Deferred.await(deferred).pipe(
893 Effect.timeoutOption(HISTORY_REQUEST_TIMEOUT),
894 Effect.flatMap(
895 Option.match({
896 onNone: () =>
897 Effect.fail(
898 new NativeTelemetryRequestTimedOut({
899 operation: "readHistory",
900 timeoutMs: Duration.toMillis(HISTORY_REQUEST_TIMEOUT),
901 }),
902 ),
903 onSome: Effect.succeed,
904 }),
905 ),
906 ),
907 ),
908 Effect.ensuring(
909 Ref.update(pendingHistories, (pending) => {
910 const next = new Map(pending);
911 next.delete(requestId);
912 return next;
913 }),
914 ),
915 );
916 });
917
918 const sampleNow: NativeTelemetryClient["Service"]["sampleNow"] = Effect.gen(function* () {
919 const current = yield* Ref.get(state);
920 if (!canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) {
921 return yield* new NativeTelemetryUnavailable({
922 reason: Option.getOrElse(current.lastError, () => "sidecar is not running"),
923 });
924 }
925
926 const requestId = yield* random.uuidV4().pipe(
927 Effect.mapError(
928 (cause) =>
929 new NativeTelemetryCommandFailed({
930 operation: "createRequestId",
931 cause,
932 }),
933 ),
934 );
935 const deferred = yield* Deferred.make<NativeTelemetrySnapshot, NativeTelemetryClientError>();
936 yield* Ref.update(pendingSamples, (pending) => {
937 const next = new Map(pending);
938 next.set(requestId, deferred);
939 return next;
940 });
941 return yield* writeCommand(Option.getOrThrow(current.handle), {
942 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
943 type: "sampleNow",
944 requestId,
945 }).pipe(
946 Effect.andThen(
947 Deferred.await(deferred).pipe(
948 Effect.timeoutOption(SAMPLE_REQUEST_TIMEOUT),
949 Effect.flatMap(
950 Option.match({
951 onNone: () =>
952 Effect.fail(
953 new NativeTelemetryRequestTimedOut({
954 operation: "sampleNow",
955 timeoutMs: Duration.toMillis(SAMPLE_REQUEST_TIMEOUT),
956 }),
957 ),
958 onSome: Effect.succeed,
959 }),
960 ),
961 ),
962 ),
963 Effect.ensuring(
964 Ref.update(pendingSamples, (pending) => {
965 const next = new Map(pending);
966 next.delete(requestId);
967 return next;
968 }),
969 ),
970 );
971 });
972
973 const processTable: NativeTelemetryClient["Service"]["processTable"] = Effect.gen(function* () {
974 const current = yield* Ref.get(state);
975 if (!canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) {
976 return yield* new NativeTelemetryUnavailable({
977 reason: Option.getOrElse(current.lastError, () => "sidecar is not running"),
978 });
979 }
980
981 const requestId = yield* random
982 .uuidV4()
983 .pipe(
984 Effect.mapError(
985 (cause) => new NativeTelemetryCommandFailed({ operation: "createRequestId", cause }),
986 ),
987 );
988 const deferred = yield* Deferred.make<
989 ReadonlyArray<ResourceMonitorProcessTableEntry>,
990 NativeTelemetryClientError
991 >();
992 yield* Ref.update(pendingProcessTables, (pending) => {
993 const next = new Map(pending);
994 next.set(requestId, deferred);
995 return next;
996 });
997 return yield* writeCommand(Option.getOrThrow(current.handle), {
998 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
999 type: "processTable",
1000 requestId,
1001 }).pipe(
1002 Effect.andThen(
1003 Deferred.await(deferred).pipe(
1004 Effect.timeoutOption(PROCESS_TABLE_REQUEST_TIMEOUT),
1005 Effect.flatMap(
1006 Option.match({
1007 onNone: () =>
1008 Effect.fail(
1009 new NativeTelemetryRequestTimedOut({
1010 operation: "processTable",
1011 timeoutMs: Duration.toMillis(PROCESS_TABLE_REQUEST_TIMEOUT),
1012 }),
1013 ),
1014 onSome: Effect.succeed,
1015 }),
1016 ),
1017 ),
1018 ),
1019 Effect.ensuring(
1020 Ref.update(pendingProcessTables, (pending) => {
1021 const next = new Map(pending);
1022 next.delete(requestId);
1023 return next;
1024 }),
1025 ),
1026 );
1027 });
1028
1029 const health = currentHealth;
1030
1031 return NativeTelemetryClient.of({
1032 capabilities: Ref.get(state).pipe(
1033 Effect.flatMap((current) =>
1034 Option.match(current.hello, {
1035 onNone: () =>
1036 Effect.fail(
1037 new NativeTelemetryUnavailable({
1038 reason: Option.getOrElse(current.lastError, () => "handshake is incomplete"),
1039 }),
1040 ),
1041 onSome: (hello) => Effect.succeed(hello.capabilities),
1042 }),
1043 ),
1044 ),
1045 snapshots: liveSnapshots,
1046 readHistory,
1047 setExternalProcesses,
1048 setHostPowerState,
1049 sampleNow,
1050 processTable,
1051 retry: Ref.get(state).pipe(
1052 Effect.flatMap((current) =>
1053 !canRequestNativeTelemetryRetry(current.status, Option.isSome(current.handle))
1054 ? Effect.succeed(false)
1055 : Queue.offer(retryQueue, undefined).pipe(Effect.as(true)),
1056 ),
1057 ),
1058 health,
1059 subscribeHealth: subscribeBeforeSnapshotWithoutMutex(healthChanges, health),
1060 });
1061});
1062
1063export const layer = Layer.effect(NativeTelemetryClient, make());
1064
1065export const layerTest = (
1066 overrides: Partial<NativeTelemetryClient["Service"]> = {},
1067): Layer.Layer<NativeTelemetryClient> => {
1068 const health =
1069 overrides.health ??
1070 Effect.succeed({
1071 status: "unavailable" as const,
1072 hello: Option.none<ResourceMonitorHelloEvent>(),
1073 lastSampleAt: Option.none<DateTime.Utc>(),
1074 lastError: Option.some("Resource monitor test implementation is unavailable."),
1075 restartCount: 0,
1076 sampleIntervalMs: IDLE_SAMPLE_INTERVAL_MS,
1077 });
1078 return Layer.succeed(
1079 NativeTelemetryClient,
1080 NativeTelemetryClient.of({
1081 capabilities: Effect.succeed({
1082 cumulativeCpuTime: true,
1083 currentCpuPercent: true,
1084 residentMemory: true,
1085 virtualMemory: true,
1086 ioBytes: true,
1087 processStartTime: true,
1088 processTree: true,
1089 }),
1090 snapshots: Stream.empty,
1091 readHistory: Effect.fn("NativeTelemetryClient.readHistory")(function () {
1092 return Effect.fail(
1093 new NativeTelemetryUnavailable({
1094 reason: "No resource monitor history was configured for this test.",
1095 }),
1096 );
1097 }),
1098 setExternalProcesses: Effect.fn("NativeTelemetryClient.setExternalProcesses")(function () {
1099 return Effect.void;
1100 }),
1101 setHostPowerState: Effect.fn("NativeTelemetryClient.setHostPowerState")(function () {
1102 return Effect.void;
1103 }),
1104 sampleNow: Effect.fail(
1105 new NativeTelemetryUnavailable({
1106 reason: "No resource monitor sample was configured for this test.",
1107 }),
1108 ),
1109 processTable: Effect.fail(
1110 new NativeTelemetryUnavailable({
1111 reason: "No resource monitor process table was configured for this test.",
1112 }),
1113 ),
1114 retry: Effect.succeed(false),
1115 health,
1116 subscribeHealth:
1117 overrides.subscribeHealth ??
1118 health.pipe(
1119 Effect.map((initial) => ({
1120 latest: initial,
1121 changes: Stream.empty,
1122 })),
1123 ),
1124 ...overrides,
1125 }),
1126 );
1127};