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 Crypto from "effect/Crypto";
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.
264export function 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
282export function commitCollectionControlUpdate<E, R>(
283 desiredState: Ref.Ref<CollectionControl>,
284 appliedState: Ref.Ref<CollectionControl>,
285 update: (current: CollectionControl) => CollectionControl,
286 apply: (previous: CollectionControl, next: CollectionControl) => Effect.Effect<void, E, R>,
287): Effect.Effect<readonly [CollectionControl, CollectionControl], E, R> {
288 return Effect.gen(function* () {
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}
299
300export function synchronizeCollectionControlOnStart<E1, R1, E2, R2>(
301 mutex: Semaphore.Semaphore,
302 desiredState: Ref.Ref<CollectionControl>,
303 appliedState: Ref.Ref<CollectionControl>,
304 apply: (control: CollectionControl) => Effect.Effect<void, E1, R1>,
305 markReady: Effect.Effect<void, E2, R2>,
306) {
307 return mutex.withPermits(1)(
308 Effect.gen(function* () {
309 const control = yield* Ref.get(desiredState);
310 yield* apply(control);
311 yield* Ref.set(appliedState, control);
312 yield* markReady;
313 return control;
314 }),
315 );
316}
317
318const decodeMonitorEvent: (
319 value: unknown,
320) => Effect.Effect<ResourceMonitorEvent, Schema.SchemaError> = Schema.decodeUnknownEffect(
321 ResourceMonitorEventSchema,
322);
323const encodeMonitorCommand = Schema.encodeEffect(
324 Schema.fromJsonString(ResourceMonitorCommandSchema),
325);
326const isProtocolMismatch = Schema.is(NativeTelemetryProtocolMismatch);
327const isDecodeFailed = Schema.is(NativeTelemetryDecodeFailed);
328const isCommandFailed = Schema.is(NativeTelemetryCommandFailed);
329
330function eventVersion(value: unknown): number | undefined {
331 if (typeof value !== "object" || value === null) return undefined;
332 const version = Reflect.get(value, "version");
333 return typeof version === "number" ? version : undefined;
334}
335
336function restartDelay(attempt: number): Duration.Duration {
337 return Duration.min(Duration.times(INITIAL_RESTART_DELAY, 2 ** attempt), MAX_RESTART_DELAY);
338}
339
340export function retainRecentNativeTelemetryFailures(
341 failures: ReadonlyArray<number>,
342 now: number,
343): ReadonlyArray<number> {
344 return failures.filter((failedAt) => now - failedAt <= FAILURE_WINDOW_MS);
345}
346
347function errorMessage(error: NativeTelemetryClientError): string {
348 return error.message;
349}
350
351export function canRequestNativeTelemetryRetry(
352 status: ResourceTelemetrySourceStatus,
353 hasHandle: boolean,
354): boolean {
355 return status !== "healthy" && status !== "starting" && !hasHandle;
356}
357
358export function canCommandNativeTelemetrySidecar(
359 status: ResourceTelemetrySourceStatus,
360 hasHandle: boolean,
361): boolean {
362 return hasHandle && (status === "healthy" || status === "degraded");
363}
364
365/** @public Service construction is part of the canonical Effect module API. */
366export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(function* () {
367 const binary = yield* ResourceMonitorBinary.ResourceMonitorBinary;
368 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
369 const crypto = yield* Crypto.Crypto;
370 const config = yield* ServerConfig.ServerConfig;
371 const initializedAt = yield* DateTime.now;
372 const state = yield* Ref.make(initialState);
373 const collectionControl = yield* Ref.make<CollectionControl>({
374 hostPower: {
375 source: "unknown",
376 idle: "unknown",
377 idleSeconds: null,
378 locked: "unknown",
379 suspended: false,
380 onBattery: "unknown",
381 lowPowerMode: "unknown",
382 thermalState: "unknown",
383 stale: true,
384 updatedAt: initializedAt,
385 },
386 liveSubscriberCount: 0,
387 sampleIntervalMs: IDLE_SAMPLE_INTERVAL_MS,
388 });
389 const appliedCollectionControl = yield* Ref.make(yield* Ref.get(collectionControl));
390 const externalProcesses = yield* Ref.make<ReadonlyArray<ResourceMonitorExternalProcess>>([]);
391 const pendingSamples = yield* Ref.make(
392 new Map<string, Deferred.Deferred<NativeTelemetrySnapshot, NativeTelemetryClientError>>(),
393 );
394 const pendingProcessTables = yield* Ref.make(
395 new Map<
396 string,
397 Deferred.Deferred<ReadonlyArray<ResourceMonitorProcessTableEntry>, NativeTelemetryClientError>
398 >(),
399 );
400 const pendingHistories = yield* Ref.make(new Map<string, PendingHistoryRequest>());
401 const snapshots = yield* PubSub.sliding<NativeTelemetrySnapshot>(8);
402 const healthChanges = yield* PubSub.sliding<NativeTelemetryClientHealth>(4);
403 const retryQueue = yield* Queue.sliding<void>(1);
404 const commandMutex = yield* Semaphore.make(1);
405 const controlMutex = yield* Semaphore.make(1);
406 const currentHealth = Effect.all([Ref.get(state), Ref.get(collectionControl)]).pipe(
407 Effect.map(([current, control]) => toHealth(current, control.sampleIntervalMs)),
408 );
409 const publishHealth = currentHealth.pipe(
410 Effect.flatMap((health) => PubSub.publish(healthChanges, health)),
411 Effect.asVoid,
412 );
413
414 const failPending = (error: NativeTelemetryClientError) =>
415 Effect.gen(function* () {
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(
426 histories.values(),
427 (request) => Deferred.fail(request.deferred, error),
428 { discard: true },
429 );
430 });
431
432 const writeCommand = (
433 handle: ChildProcessSpawner.ChildProcessHandle,
434 command: ResourceMonitorCommand,
435 ): Effect.Effect<void, NativeTelemetryClientError> =>
436 commandMutex.withPermits(1)(
437 encodeMonitorCommand(command).pipe(
438 Effect.map((encoded) => `${encoded}\n`),
439 Effect.mapError(
440 (cause) =>
441 new NativeTelemetryCommandFailed({
442 operation: command.type,
443 cause,
444 }),
445 ),
446 Effect.flatMap((encoded) =>
447 Stream.run(Stream.encodeText(Stream.make(encoded)), handle.stdin),
448 ),
449 Effect.mapError(
450 (cause) =>
451 new NativeTelemetryCommandFailed({
452 operation: command.type,
453 cause,
454 }),
455 ),
456 ),
457 );
458
459 const processEvent = (
460 event: ResourceMonitorEvent,
461 helloDeferred: Deferred.Deferred<ResourceMonitorHelloEvent>,
462 generation: number,
463 ): Effect.Effect<void, NativeTelemetryClientError> => {
464 switch (event.type) {
465 case "hello":
466 return Ref.update(state, (current) => ({
467 ...current,
468 status: "starting" as const,
469 hello: Option.some(event),
470 lastError: Option.none(),
471 })).pipe(
472 Effect.andThen(publishHealth),
473 Effect.andThen(Deferred.succeed(helloDeferred, event)),
474 Effect.asVoid,
475 );
476 case "snapshot":
477 return Effect.gen(function* () {
478 const nativeSnapshot = { generation, snapshot: event } satisfies NativeTelemetrySnapshot;
479 const sampledAt = DateTime.makeUnsafe(event.sampledAtUnixMs);
480 const healthChanged = yield* Ref.modify(state, (current) => [
481 current.status !== "healthy" || Option.isSome(current.lastError),
482 {
483 ...current,
484 status: "healthy" as const,
485 lastSampleAt: Option.some(sampledAt),
486 lastError: Option.none(),
487 },
488 ]);
489 if (healthChanged) yield* publishHealth;
490 yield* PubSub.publish(snapshots, nativeSnapshot);
491 if (event.requestId) {
492 const deferred = yield* Ref.modify(pendingSamples, (pending) => {
493 const next = new Map(pending);
494 const current = next.get(event.requestId!);
495 next.delete(event.requestId!);
496 return [Option.fromUndefinedOr(current), next];
497 });
498 if (Option.isSome(deferred)) {
499 yield* Deferred.succeed(deferred.value, nativeSnapshot);
500 }
501 }
502 });
503 case "processTable":
504 return Ref.modify(pendingProcessTables, (pending) => {
505 const next = new Map(pending);
506 const deferred = next.get(event.requestId);
507 next.delete(event.requestId);
508 return [Option.fromUndefinedOr(deferred), next] as const;
509 }).pipe(
510 Effect.flatMap(
511 Option.match({
512 onNone: () => Effect.void,
513 onSome: (deferred) => Deferred.succeed(deferred, event.processes),
514 }),
515 ),
516 Effect.asVoid,
517 );
518 case "historyChunk":
519 return Effect.gen(function* () {
520 const latestSnapshot = event.snapshots.at(-1);
521 const healthChanged = yield* Ref.modify(state, (current) => [
522 current.status !== "healthy" || Option.isSome(current.lastError),
523 {
524 ...current,
525 status: "healthy" as const,
526 lastSampleAt: latestSnapshot
527 ? Option.some(DateTime.makeUnsafe(latestSnapshot.sampledAtUnixMs))
528 : current.lastSampleAt,
529 lastError: Option.none(),
530 },
531 ]);
532 if (healthChanged) yield* publishHealth;
533 const completed = yield* Ref.modify(pendingHistories, (pending) => {
534 const request = pending.get(event.requestId);
535 if (!request) return [Option.none(), pending] as const;
536 const snapshots = [...request.snapshots, ...event.snapshots];
537 const next = new Map(pending);
538 if (event.done) {
539 next.delete(event.requestId);
540 return [Option.some({ deferred: request.deferred, snapshots }), next] as const;
541 }
542 next.set(event.requestId, { deferred: request.deferred, snapshots });
543 return [Option.none(), next] as const;
544 });
545 if (Option.isSome(completed)) {
546 yield* Deferred.succeed(completed.value.deferred, completed.value.snapshots);
547 }
548 });
549 case "error":
550 return Ref.update(state, (current) => ({
551 ...current,
552 status: "degraded" as const,
553 lastError: Option.some(event.message),
554 })).pipe(
555 Effect.andThen(publishHealth),
556 Effect.andThen(
557 event.recoverable
558 ? Effect.void
559 : Effect.fail(
560 new NativeTelemetryCommandFailed({
561 operation: event.code,
562 cause: event.message,
563 }),
564 ),
565 ),
566 );
567 }
568 };
569
570 const runAttempt: Effect.Effect<void, NativeTelemetryClientError> = Effect.scoped(
571 Effect.gen(function* () {
572 const executablePath = yield* binary.resolve;
573 const command = ChildProcess.make(executablePath, [], {
574 cwd: config.cwd,
575 stdin: {
576 stream: "pipe",
577 endOnDone: false,
578 },
579 stdout: "pipe",
580 stderr: "pipe",
581 killSignal: "SIGTERM",
582 forceKillAfter: Duration.seconds(2),
583 });
584 const handle = yield* Effect.acquireRelease(
585 spawner
586 .spawn(command)
587 .pipe(
588 Effect.mapError(
589 (cause) => new NativeTelemetrySpawnFailed({ path: executablePath, cause }),
590 ),
591 ),
592 (child) => child.kill().pipe(Effect.ignore),
593 );
594 yield* Ref.update(state, (current) => ({
595 ...current,
596 status: "starting" as const,
597 handle: Option.some(handle),
598 hello: Option.none(),
599 }));
600 yield* publishHealth;
601 const generation = (yield* Ref.get(state)).restartCount;
602
603 const helloDeferred = yield* Deferred.make<ResourceMonitorHelloEvent>();
604 const eventFiber = yield* handle.stdout.pipe(
605 Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),
606 Stream.mapEffect(
607 (
608 value,
609 ): Effect.Effect<
610 ResourceMonitorEvent,
611 NativeTelemetryProtocolMismatch | NativeTelemetryDecodeFailed
612 > => {
613 const version = eventVersion(value);
614 if (version !== undefined && version !== RESOURCE_MONITOR_PROTOCOL_VERSION) {
615 return Effect.fail(
616 new NativeTelemetryProtocolMismatch({
617 expectedVersion: RESOURCE_MONITOR_PROTOCOL_VERSION,
618 receivedVersion: version,
619 }),
620 );
621 }
622 return decodeMonitorEvent(value).pipe(
623 Effect.mapError((cause) => new NativeTelemetryDecodeFailed({ cause })),
624 );
625 },
626 ),
627 Stream.runForEach((event) => processEvent(event, helloDeferred, generation)),
628 Effect.mapError((cause) =>
629 isProtocolMismatch(cause) || isDecodeFailed(cause) || isCommandFailed(cause)
630 ? cause
631 : new NativeTelemetryDecodeFailed({ cause }),
632 ),
633 Effect.forkScoped,
634 );
635 yield* handle.stderr.pipe(Stream.runDrain, Effect.ignore, Effect.forkScoped);
636
637 const hello = yield* Deferred.await(helloDeferred).pipe(
638 Effect.timeoutOption(HANDSHAKE_TIMEOUT),
639 Effect.flatMap(
640 Option.match({
641 onNone: () =>
642 Effect.fail(
643 new NativeTelemetryHandshakeTimedOut({
644 timeoutMs: Duration.toMillis(HANDSHAKE_TIMEOUT),
645 }),
646 ),
647 onSome: Effect.succeed,
648 }),
649 ),
650 );
651 yield* synchronizeCollectionControlOnStart(
652 controlMutex,
653 collectionControl,
654 appliedCollectionControl,
655 (control) =>
656 Effect.gen(function* () {
657 yield* writeCommand(handle, {
658 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
659 type: "configure",
660 rootPid: process.pid,
661 sampleIntervalMs: control.sampleIntervalMs,
662 externalProcesses: [...(yield* Ref.get(externalProcesses))],
663 });
664 if (control.liveSubscriberCount > 0) {
665 yield* writeCommand(handle, {
666 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
667 type: "setStreaming",
668 enabled: true,
669 });
670 }
671 }),
672 Ref.update(state, (current) => ({
673 ...current,
674 status: "healthy" as const,
675 hello: Option.some(hello),
676 })).pipe(Effect.andThen(publishHealth)),
677 );
678
679 yield* writeCommand(handle, {
680 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
681 type: "setExternalProcesses",
682 processes: [...(yield* Ref.get(externalProcesses))],
683 });
684
685 const exitEffect = handle.exitCode.pipe(
686 Effect.mapError(
687 (cause) =>
688 new NativeTelemetryCommandFailed({
689 operation: "waitForExit",
690 cause,
691 }),
692 ),
693 Effect.flatMap((exitCode) =>
694 Effect.fail(new NativeTelemetryExited({ exitCode: Number(exitCode) })),
695 ),
696 );
697 const decoderEffect = Fiber.join(eventFiber).pipe(
698 Effect.andThen(Effect.fail(new NativeTelemetryStreamClosed())),
699 );
700 return yield* Effect.raceFirst(exitEffect, decoderEffect);
701 }),
702 ).pipe(
703 Effect.ensuring(
704 Ref.update(state, (current) => ({
705 ...current,
706 handle: Option.none(),
707 })),
708 ),
709 );
710
711 yield* Effect.gen(function* () {
712 let failures: ReadonlyArray<number> = [];
713 let restartAttempt = 0;
714
715 while (true) {
716 const result = yield* Effect.result(runAttempt);
717 if (Result.isSuccess(result)) {
718 return;
719 }
720
721 const error = result.failure;
722 const now = DateTime.toEpochMillis(yield* DateTime.now);
723 const recentFailures = retainRecentNativeTelemetryFailures(failures, now);
724 if (recentFailures.length === 0) {
725 restartAttempt = 0;
726 }
727 failures = [...recentFailures, now];
728 const exhausted = failures.length >= MAX_FAILURES_PER_WINDOW;
729 yield* Ref.update(state, (current) => ({
730 ...current,
731 status: exhausted ? ("unavailable" as const) : ("degraded" as const),
732 hello: Option.none(),
733 lastError: Option.some(errorMessage(error)),
734 restartCount: current.restartCount + 1,
735 }));
736 yield* publishHealth;
737 yield* failPending(error);
738
739 if (exhausted) {
740 yield* Queue.take(retryQueue);
741 failures = [];
742 restartAttempt = 0;
743 yield* Ref.update(state, (current) => ({
744 ...current,
745 status: "starting" as const,
746 hello: Option.none(),
747 lastError: Option.none(),
748 }));
749 yield* publishHealth;
750 continue;
751 }
752
753 const manuallyRetried = yield* Effect.raceFirst(
754 Effect.sleep(restartDelay(restartAttempt)).pipe(Effect.as(false)),
755 Queue.take(retryQueue).pipe(Effect.as(true)),
756 );
757 restartAttempt = manuallyRetried ? 0 : restartAttempt + 1;
758 }
759 }).pipe(
760 Effect.catchCause((cause) =>
761 Cause.hasInterruptsOnly(cause)
762 ? Effect.void
763 : Ref.update(state, (current) => ({
764 ...current,
765 status: "unavailable" as const,
766 hello: Option.none(),
767 lastError: Option.some("Resource monitor supervisor stopped unexpectedly."),
768 })).pipe(
769 Effect.andThen(publishHealth),
770 Effect.andThen(
771 Effect.logWarning("Resource monitor supervisor failed", {
772 cause: Cause.pretty(cause),
773 }),
774 ),
775 ),
776 ),
777 Effect.forkScoped,
778 );
779
780 const applyCollectionControl = Effect.fn(
781 "resourceTelemetry.nativeTelemetryClient.applyCollectionControl",
782 )(function* (previous: CollectionControl, next: CollectionControl) {
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 const setExternalProcesses: NativeTelemetryClient["Service"]["setExternalProcesses"] = (
846 processes,
847 ) =>
848 Effect.gen(function* () {
849 yield* Ref.set(externalProcesses, [...processes]);
850 const current = yield* Ref.get(state);
851 if (!canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) return;
852 yield* writeCommand(Option.getOrThrow(current.handle), {
853 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
854 type: "setExternalProcesses",
855 processes: [...processes],
856 });
857 });
858
859 const readHistory: NativeTelemetryClient["Service"]["readHistory"] = (windowMs) =>
860 Effect.gen(function* () {
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* crypto.randomUUIDv4().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* crypto.randomUUIDv4().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* crypto
982 .randomUUIDv4()
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: () =>
1092 Effect.fail(
1093 new NativeTelemetryUnavailable({
1094 reason: "No resource monitor history was configured for this test.",
1095 }),
1096 ),
1097 setExternalProcesses: () => Effect.void,
1098 setHostPowerState: () => Effect.void,
1099 sampleNow: Effect.fail(
1100 new NativeTelemetryUnavailable({
1101 reason: "No resource monitor sample was configured for this test.",
1102 }),
1103 ),
1104 processTable: Effect.fail(
1105 new NativeTelemetryUnavailable({
1106 reason: "No resource monitor process table was configured for this test.",
1107 }),
1108 ),
1109 retry: Effect.succeed(false),
1110 health,
1111 subscribeHealth:
1112 overrides.subscribeHealth ??
1113 health.pipe(
1114 Effect.map((initial) => ({
1115 latest: initial,
1116 changes: Stream.empty,
1117 })),
1118 ),
1119 ...overrides,
1120 }),
1121 );
1122};