Skip to content

Commit da075ee

Browse files
juliusmarmingecodex
juliusmarminge
and
codex
committed
Align diagnostics and telemetry Effect services
Co-authored-by: codex <[email protected]>
1 parent 01492eb commit da075ee

18 files changed

Lines changed: 461 additions & 299 deletions

‎apps/server/src/diagnostics/ProcessDiagnostics.ts‎

Lines changed: 103 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,8 @@ import * as Effect from "effect/Effect";
1212
import * as Layer from "effect/Layer";
1313
import * as Option from "effect/Option";
1414
import * as Schema from "effect/Schema";
15-
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
15+
import * as ChildProcess from "effect/unstable/process/ChildProcess";
16+
import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
1617

1718
import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts";
1819

@@ -31,35 +32,80 @@ const PROCESS_QUERY_TIMEOUT_MS = 1_000;
3132
const POSIX_PROCESS_QUERY_COMMAND = "pid=,ppid=,pgid=,stat=,pcpu=,rss=,etime=,command=";
3233
const PROCESS_QUERY_MAX_OUTPUT_BYTES = 2 * 1024 * 1024;
3334

34-
export interface ProcessDiagnosticsShape {
35-
readonly read: Effect.Effect<ServerProcessDiagnosticsResult>;
36-
readonly signal: (input: {
37-
readonly pid: number;
38-
readonly signal: ServerProcessSignal;
39-
}) => Effect.Effect<ServerSignalProcessResult>;
40-
}
41-
4235
export class ProcessDiagnostics extends Context.Service<
4336
ProcessDiagnostics,
44-
ProcessDiagnosticsShape
37+
{
38+
readonly read: Effect.Effect<ServerProcessDiagnosticsResult>;
39+
readonly signal: (input: {
40+
readonly pid: number;
41+
readonly signal: ServerProcessSignal;
42+
}) => Effect.Effect<ServerSignalProcessResult>;
43+
}
4544
>()("t3/diagnostics/ProcessDiagnostics") {}
4645

47-
class ProcessDiagnosticsError extends Schema.TaggedErrorClass<ProcessDiagnosticsError>()(
48-
"ProcessDiagnosticsError",
46+
class ProcessDiagnosticsQueryTimeoutError extends Schema.TaggedErrorClass<ProcessDiagnosticsQueryTimeoutError>()(
47+
"ProcessDiagnosticsQueryTimeoutError",
48+
{ command: Schema.String },
49+
) {
50+
override get message(): string {
51+
return `Process diagnostics query '${this.command}' timed out.`;
52+
}
53+
}
54+
55+
class ProcessDiagnosticsQueryFailedError extends Schema.TaggedErrorClass<ProcessDiagnosticsQueryFailedError>()(
56+
"ProcessDiagnosticsQueryFailedError",
4957
{
50-
message: Schema.String,
58+
command: Schema.String,
59+
stderr: Schema.optional(Schema.String),
5160
cause: Schema.optional(Schema.Defect()),
5261
},
53-
) {}
54-
const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError);
62+
) {
63+
override get message(): string {
64+
return this.stderr?.trim() || `Failed to query process diagnostics with '${this.command}'.`;
65+
}
66+
}
5567

56-
function toProcessDiagnosticsError(message: string, cause?: unknown): ProcessDiagnosticsError {
57-
return new ProcessDiagnosticsError({
58-
message,
59-
...(cause === undefined ? {} : { cause }),
60-
});
68+
class ProcessDiagnosticsServerProcessSignalError extends Schema.TaggedErrorClass<ProcessDiagnosticsServerProcessSignalError>()(
69+
"ProcessDiagnosticsServerProcessSignalError",
70+
{ pid: Schema.Number },
71+
) {
72+
override get message(): string {
73+
return "Refusing to signal the T3 server process.";
74+
}
75+
}
76+
77+
class ProcessDiagnosticsNotDescendantError extends Schema.TaggedErrorClass<ProcessDiagnosticsNotDescendantError>()(
78+
"ProcessDiagnosticsNotDescendantError",
79+
{ pid: Schema.Number },
80+
) {
81+
override get message(): string {
82+
return `Process ${this.pid} is not a live descendant of the T3 server.`;
83+
}
6184
}
6285

86+
class ProcessDiagnosticsSignalFailedError extends Schema.TaggedErrorClass<ProcessDiagnosticsSignalFailedError>()(
87+
"ProcessDiagnosticsSignalFailedError",
88+
{
89+
pid: Schema.Number,
90+
signal: Schema.String,
91+
cause: Schema.Defect(),
92+
},
93+
) {
94+
override get message(): string {
95+
return `Failed to signal process ${this.pid} with ${this.signal}.`;
96+
}
97+
}
98+
99+
const ProcessDiagnosticsError = Schema.Union([
100+
ProcessDiagnosticsQueryTimeoutError,
101+
ProcessDiagnosticsQueryFailedError,
102+
ProcessDiagnosticsServerProcessSignalError,
103+
ProcessDiagnosticsNotDescendantError,
104+
ProcessDiagnosticsSignalFailedError,
105+
]);
106+
type ProcessDiagnosticsError = typeof ProcessDiagnosticsError.Type;
107+
const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError);
108+
63109
function parsePositiveInt(value: string): number | null {
64110
const parsed = Number.parseInt(value, 10);
65111
return Number.isInteger(parsed) && parsed > 0 ? parsed : null;
@@ -272,11 +318,7 @@ interface ProcessOutput {
272318
}
273319

274320
const runProcess = Effect.fn("runProcess")(
275-
function* (input: {
276-
readonly command: string;
277-
readonly args: ReadonlyArray<string>;
278-
readonly errorMessage: string;
279-
}) {
321+
function* (input: { readonly command: string; readonly args: ReadonlyArray<string> }) {
280322
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
281323
// `ps` and `powershell.exe` are real executables; spawning through cmd.exe
282324
// shell mode would re-tokenize the PowerShell `-Command` payload (which
@@ -315,14 +357,22 @@ const runProcess = Effect.fn("runProcess")(
315357
Effect.timeoutOption(Duration.millis(PROCESS_QUERY_TIMEOUT_MS)),
316358
Effect.flatMap((result) =>
317359
Option.match(result, {
318-
onNone: () => Effect.fail(toProcessDiagnosticsError(`${input.errorMessage} timed out.`)),
360+
onNone: () =>
361+
Effect.fail(
362+
new ProcessDiagnosticsQueryTimeoutError({
363+
command: input.command,
364+
}),
365+
),
319366
onSome: Effect.succeed,
320367
}),
321368
),
322369
Effect.mapError((cause) =>
323370
isProcessDiagnosticsError(cause)
324371
? cause
325-
: toProcessDiagnosticsError(input.errorMessage, cause),
372+
: new ProcessDiagnosticsQueryFailedError({
373+
command: input.command,
374+
cause,
375+
}),
326376
),
327377
),
328378
);
@@ -335,11 +385,15 @@ function readPosixProcessRows(): Effect.Effect<
335385
return runProcess({
336386
command: "ps",
337387
args: ["-axo", POSIX_PROCESS_QUERY_COMMAND],
338-
errorMessage: "Failed to query process diagnostics.",
339388
}).pipe(
340389
Effect.flatMap((result) =>
341390
result.exitCode !== 0
342-
? Effect.fail(toProcessDiagnosticsError(result.stderr.trim() || "ps failed."))
391+
? Effect.fail(
392+
new ProcessDiagnosticsQueryFailedError({
393+
command: "ps",
394+
stderr: result.stderr.trim() || "ps failed.",
395+
}),
396+
)
343397
: Effect.succeed(parsePosixProcessRows(result.stdout)),
344398
),
345399
);
@@ -361,12 +415,14 @@ function readWindowsProcessRows(): Effect.Effect<
361415
return runProcess({
362416
command: "powershell.exe",
363417
args: ["-NoProfile", "-NonInteractive", "-Command", command],
364-
errorMessage: "Failed to query process diagnostics.",
365418
}).pipe(
366419
Effect.flatMap((result) =>
367420
result.exitCode !== 0
368421
? Effect.fail(
369-
toProcessDiagnosticsError(result.stderr.trim() || "PowerShell process query failed."),
422+
new ProcessDiagnosticsQueryFailedError({
423+
command: "powershell.exe",
424+
stderr: result.stderr.trim() || "PowerShell process query failed.",
425+
}),
370426
)
371427
: Effect.succeed(parseWindowsProcessRows(result.stdout)),
372428
),
@@ -390,7 +446,11 @@ function assertDescendantPid(
390446
pid: number,
391447
): Effect.Effect<void, ProcessDiagnosticsError, ChildProcessSpawner.ChildProcessSpawner> {
392448
if (pid === process.pid) {
393-
return Effect.fail(toProcessDiagnosticsError("Refusing to signal the T3 server process."));
449+
return Effect.fail(
450+
new ProcessDiagnosticsServerProcessSignalError({
451+
pid,
452+
}),
453+
);
394454
}
395455

396456
return readProcessRows.pipe(
@@ -402,16 +462,18 @@ function assertDescendantPid(
402462
return descendant
403463
? Effect.void
404464
: Effect.fail(
405-
toProcessDiagnosticsError(`Process ${pid} is not a live descendant of the T3 server.`),
465+
new ProcessDiagnosticsNotDescendantError({
466+
pid,
467+
}),
406468
);
407469
}),
408470
);
409471
}
410472

411-
export const make = Effect.fn("makeProcessDiagnostics")(function* () {
473+
export const make = Effect.gen(function* () {
412474
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
413475

414-
const read: ProcessDiagnosticsShape["read"] = Effect.gen(function* () {
476+
const read: ProcessDiagnostics["Service"]["read"] = Effect.gen(function* () {
415477
const readAt = yield* DateTime.now;
416478
const rows = yield* readProcessRows.pipe(
417479
Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
@@ -427,7 +489,7 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
427489
),
428490
);
429491

430-
const signal: ProcessDiagnosticsShape["signal"] = Effect.fn("ProcessDiagnostics.signal")(
492+
const signal: ProcessDiagnostics["Service"]["signal"] = Effect.fn("ProcessDiagnostics.signal")(
431493
function* (input) {
432494
return yield* assertDescendantPid(input.pid).pipe(
433495
Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
@@ -443,10 +505,11 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
443505
};
444506
},
445507
catch: (cause) =>
446-
toProcessDiagnosticsError(
447-
`Failed to signal process ${input.pid} with ${input.signal}.`,
508+
new ProcessDiagnosticsSignalFailedError({
509+
pid: input.pid,
510+
signal: input.signal,
448511
cause,
449-
),
512+
}),
450513
}),
451514
),
452515
Effect.catch((error: ProcessDiagnosticsError) =>
@@ -464,4 +527,4 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
464527
return ProcessDiagnostics.of({ read, signal });
465528
});
466529

467-
export const layer = Layer.effect(ProcessDiagnostics, make());
530+
export const layer = Layer.effect(ProcessDiagnostics, make);

‎apps/server/src/diagnostics/ProcessResourceMonitor.test.ts‎

Lines changed: 10 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -3,16 +3,13 @@ import * as DateTime from "effect/DateTime";
33
import * as Effect from "effect/Effect";
44
import * as Option from "effect/Option";
55

6-
import {
7-
aggregateProcessResourceHistory,
8-
collectMonitoredSamples,
9-
} from "./ProcessResourceMonitor.ts";
6+
import * as ProcessResourceMonitor from "./ProcessResourceMonitor.ts";
107

118
describe("ProcessResourceMonitor", () => {
129
it.effect("samples the server root process and descendants", () =>
1310
Effect.sync(() => {
1411
const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
15-
const samples = collectMonitoredSamples({
12+
const samples = ProcessResourceMonitor.collectMonitoredSamples({
1613
serverPid: 100,
1714
sampledAt,
1815
sampledAtMs: DateTime.toEpochMillis(sampledAt),
@@ -72,7 +69,7 @@ describe("ProcessResourceMonitor", () => {
7269
const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
7370
const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.000Z");
7471
const samples = [
75-
...collectMonitoredSamples({
72+
...ProcessResourceMonitor.collectMonitoredSamples({
7673
serverPid: 100,
7774
sampledAt: firstAt,
7875
sampledAtMs: DateTime.toEpochMillis(firstAt),
@@ -89,7 +86,7 @@ describe("ProcessResourceMonitor", () => {
8986
},
9087
],
9188
}),
92-
...collectMonitoredSamples({
89+
...ProcessResourceMonitor.collectMonitoredSamples({
9390
serverPid: 100,
9491
sampledAt: secondAt,
9592
sampledAtMs: DateTime.toEpochMillis(secondAt),
@@ -108,7 +105,7 @@ describe("ProcessResourceMonitor", () => {
108105
}),
109106
];
110107

111-
const result = aggregateProcessResourceHistory({
108+
const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
112109
samples,
113110
readAt: secondAt,
114111
readAtMs: DateTime.toEpochMillis(secondAt),
@@ -132,7 +129,7 @@ describe("ProcessResourceMonitor", () => {
132129
const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.400Z");
133130
const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.900Z");
134131
const samples = [
135-
...collectMonitoredSamples({
132+
...ProcessResourceMonitor.collectMonitoredSamples({
136133
serverPid: 100,
137134
sampledAt: firstAt,
138135
sampledAtMs: DateTime.toEpochMillis(firstAt),
@@ -149,7 +146,7 @@ describe("ProcessResourceMonitor", () => {
149146
},
150147
],
151148
}),
152-
...collectMonitoredSamples({
149+
...ProcessResourceMonitor.collectMonitoredSamples({
153150
serverPid: 100,
154151
sampledAt: secondAt,
155152
sampledAtMs: DateTime.toEpochMillis(secondAt),
@@ -168,7 +165,7 @@ describe("ProcessResourceMonitor", () => {
168165
}),
169166
];
170167

171-
const result = aggregateProcessResourceHistory({
168+
const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
172169
samples,
173170
readAt: secondAt,
174171
readAtMs: DateTime.toEpochMillis(secondAt),
@@ -187,7 +184,7 @@ describe("ProcessResourceMonitor", () => {
187184
it.effect("returns all process summaries in the selected window", () =>
188185
Effect.sync(() => {
189186
const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
190-
const samples = collectMonitoredSamples({
187+
const samples = ProcessResourceMonitor.collectMonitoredSamples({
191188
serverPid: 100,
192189
sampledAt,
193190
sampledAtMs: DateTime.toEpochMillis(sampledAt),
@@ -215,7 +212,7 @@ describe("ProcessResourceMonitor", () => {
215212
],
216213
});
217214

218-
const result = aggregateProcessResourceHistory({
215+
const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
219216
samples,
220217
readAt: sampledAt,
221218
readAtMs: DateTime.toEpochMillis(sampledAt),

0 commit comments

Comments
 (0)