Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/server/src/diagnostics/ProcessDiagnostics.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ function makeNativeSnapshot(
processes: ResourceMonitorSnapshotEvent["processes"],
): ResourceMonitorSnapshotEvent {
return {
version: 2,
version: 3,
type: "snapshot",
sequence: 1,
sampledAtUnixMs: DateTime.toEpochMillis(DateTime.makeUnsafe("2026-05-05T10:00:00.000Z")),
Expand Down
2 changes: 1 addition & 1 deletion apps/server/src/resourceTelemetry/Model.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ function nativeSnapshot(
sequence = 1,
): ResourceMonitorSnapshotEvent {
return {
version: 2,
version: 3,
type: "snapshot",
sequence,
sampledAtUnixMs,
Expand Down
10 changes: 9 additions & 1 deletion apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ describe("canCommandNativeTelemetrySidecar", () => {
});

describe("NativeTelemetryRequestTimedOut", () => {
it("models history and sample request deadlines without a fabricated cause", () => {
it("models request deadlines without a fabricated cause", () => {
const historyTimeout = new NativeTelemetryRequestTimedOut({
operation: "readHistory",
timeoutMs: 15_000,
Expand All @@ -92,15 +92,23 @@ describe("NativeTelemetryRequestTimedOut", () => {
operation: "sampleNow",
timeoutMs: 5_000,
});
const processTableTimeout = new NativeTelemetryRequestTimedOut({
operation: "processTable",
timeoutMs: 5_000,
});

expect(historyTimeout.message).toBe(
"Resource monitor 'readHistory' request timed out after 15000ms.",
);
expect(sampleTimeout.message).toBe(
"Resource monitor 'sampleNow' request timed out after 5000ms.",
);
expect(processTableTimeout.message).toBe(
"Resource monitor 'processTable' request timed out after 5000ms.",
);
expect("cause" in historyTimeout).toBe(false);
expect("cause" in sampleTimeout).toBe(false);
expect("cause" in processTableTimeout).toBe(false);
});
});

Expand Down
93 changes: 92 additions & 1 deletion apps/server/src/resourceTelemetry/NativeTelemetryClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import type {
ResourceMonitorEvent,
ResourceMonitorExternalProcess,
ResourceMonitorHelloEvent,
ResourceMonitorProcessTableEntry,
ResourceMonitorSnapshotEvent,
ResourceTelemetrySourceStatus,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -44,6 +45,7 @@ const BATTERY_SAMPLE_INTERVAL_MS = 5_000;
const CONSTRAINED_SAMPLE_INTERVAL_MS = 15_000;
const HANDSHAKE_TIMEOUT = Duration.seconds(5);
const SAMPLE_REQUEST_TIMEOUT = Duration.seconds(5);
const PROCESS_TABLE_REQUEST_TIMEOUT = Duration.seconds(5);
const HISTORY_REQUEST_TIMEOUT = Duration.seconds(15);
const INITIAL_RESTART_DELAY = Duration.millis(500);
const MAX_RESTART_DELAY = Duration.seconds(10);
Expand Down Expand Up @@ -76,7 +78,7 @@ export class NativeTelemetryHandshakeTimedOut extends Schema.TaggedErrorClass<Na
export class NativeTelemetryRequestTimedOut extends Schema.TaggedErrorClass<NativeTelemetryRequestTimedOut>()(
"NativeTelemetryRequestTimedOut",
{
operation: Schema.Literals(["readHistory", "sampleNow"]),
operation: Schema.Literals(["processTable", "readHistory", "sampleNow"]),
timeoutMs: Schema.Number,
},
) {
Expand Down Expand Up @@ -192,6 +194,10 @@ export class NativeTelemetryClient extends Context.Service<
snapshot: HostPowerSnapshot,
) => Effect.Effect<void, NativeTelemetryClientError>;
readonly sampleNow: Effect.Effect<NativeTelemetrySnapshot, NativeTelemetryClientError>;
readonly processTable: Effect.Effect<
ReadonlyArray<ResourceMonitorProcessTableEntry>,
NativeTelemetryClientError
>;
readonly retry: Effect.Effect<boolean>;
readonly health: Effect.Effect<NativeTelemetryClientHealth>;
readonly subscribeHealth: Effect.Effect<
Expand Down Expand Up @@ -386,6 +392,12 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
const pendingSamples = yield* Ref.make(
new Map<string, Deferred.Deferred<NativeTelemetrySnapshot, NativeTelemetryClientError>>(),
);
const pendingProcessTables = yield* Ref.make(
new Map<
string,
Deferred.Deferred<ReadonlyArray<ResourceMonitorProcessTableEntry>, NativeTelemetryClientError>
>(),
);
const pendingHistories = yield* Ref.make(new Map<string, PendingHistoryRequest>());
const snapshots = yield* PubSub.sliding<NativeTelemetrySnapshot>(8);
const healthChanges = yield* PubSub.sliding<NativeTelemetryClientHealth>(4);
Expand All @@ -403,10 +415,14 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
const failPending = (error: NativeTelemetryClientError) =>
Effect.gen(function* () {
const samples = yield* Ref.getAndSet(pendingSamples, new Map());
const processTables = yield* Ref.getAndSet(pendingProcessTables, new Map());
const histories = yield* Ref.getAndSet(pendingHistories, new Map());
yield* Effect.forEach(samples.values(), (deferred) => Deferred.fail(deferred, error), {
discard: true,
});
yield* Effect.forEach(processTables.values(), (deferred) => Deferred.fail(deferred, error), {
discard: true,
});
yield* Effect.forEach(
histories.values(),
(request) => Deferred.fail(request.deferred, error),
Expand Down Expand Up @@ -485,6 +501,21 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
}
}
});
case "processTable":
return Ref.modify(pendingProcessTables, (pending) => {
const next = new Map(pending);
const deferred = next.get(event.requestId);
next.delete(event.requestId);
return [Option.fromUndefinedOr(deferred), next] as const;
}).pipe(
Effect.flatMap(
Option.match({
onNone: () => Effect.void,
onSome: (deferred) => Deferred.succeed(deferred, event.processes),
}),
),
Effect.asVoid,
);
case "historyChunk":
return Effect.gen(function* () {
const latestSnapshot = event.snapshots.at(-1);
Expand Down Expand Up @@ -940,6 +971,60 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
);
});

const processTable: NativeTelemetryClient["Service"]["processTable"] = Effect.gen(function* () {
const current = yield* Ref.get(state);
if (!canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) {
return yield* new NativeTelemetryUnavailable({
reason: Option.getOrElse(current.lastError, () => "sidecar is not running"),
});
}

const requestId = yield* crypto.randomUUIDv4.pipe(
Effect.mapError(
(cause) => new NativeTelemetryCommandFailed({ operation: "createRequestId", cause }),
),
);
const deferred = yield* Deferred.make<
ReadonlyArray<ResourceMonitorProcessTableEntry>,
NativeTelemetryClientError
>();
yield* Ref.update(pendingProcessTables, (pending) => {
const next = new Map(pending);
next.set(requestId, deferred);
return next;
});
return yield* writeCommand(Option.getOrThrow(current.handle), {
version: RESOURCE_MONITOR_PROTOCOL_VERSION,
type: "processTable",
requestId,
}).pipe(
Effect.andThen(
Deferred.await(deferred).pipe(
Effect.timeoutOption(PROCESS_TABLE_REQUEST_TIMEOUT),
Effect.flatMap(
Option.match({
onNone: () =>
Effect.fail(
new NativeTelemetryRequestTimedOut({
operation: "processTable",
timeoutMs: Duration.toMillis(PROCESS_TABLE_REQUEST_TIMEOUT),
}),
),
onSome: Effect.succeed,
}),
),
),
),
Effect.ensuring(
Ref.update(pendingProcessTables, (pending) => {
const next = new Map(pending);
next.delete(requestId);
return next;
}),
),
);
});

const health = currentHealth;

return NativeTelemetryClient.of({
Expand All @@ -961,6 +1046,7 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
setExternalProcesses,
setHostPowerState,
sampleNow,
processTable,
retry: Ref.get(state).pipe(
Effect.flatMap((current) =>
!canRequestNativeTelemetryRetry(current.status, Option.isSome(current.handle))
Expand Down Expand Up @@ -1014,6 +1100,11 @@ export const layerTest = (
reason: "No resource monitor sample was configured for this test.",
}),
),
processTable: Effect.fail(
new NativeTelemetryUnavailable({
reason: "No resource monitor process table was configured for this test.",
}),
),
retry: Effect.succeed(false),
health,
subscribeHealth:
Expand Down
4 changes: 2 additions & 2 deletions apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ function nativeSnapshot(input: {
}),
];
return {
version: 2,
version: 3,
type: "snapshot",
sequence: input.sequence,
sampledAtUnixMs: input.sampledAtUnixMs,
Expand Down Expand Up @@ -497,7 +497,7 @@ describe("ResourceTelemetry", () => {
const nativeHealth = yield* Ref.make<NativeTelemetryClient.NativeTelemetryClientHealth>({
status: "healthy",
hello: Option.some({
version: 2,
version: 3,
type: "hello",
sidecarVersion: "0.1.0",
sidecarPid: 9_000,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ function snapshot(
}),
];
return {
version: 2,
version: 3,
type: "snapshot",
sequence,
sampledAtUnixMs,
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,7 @@ const PortScannerLayerLive = PortScanner.layer.pipe(Layer.provide(ProcessRunner.
const TerminalLayerLive = TerminalManager.layer.pipe(
Layer.provide(PtyAdapterLive),
Layer.provide(PortScannerLayerLive),
Layer.provide(NativeTelemetryLayerLive),
);

const PreviewLayerLive = Layer.empty.pipe(
Expand Down
96 changes: 96 additions & 0 deletions apps/server/src/terminal/Manager.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import {
} from "@t3tools/contracts";
import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
import * as Data from "effect/Data";
import * as Clock from "effect/Clock";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Encoding from "effect/Encoding";
Expand Down Expand Up @@ -210,6 +211,10 @@ interface CreateManagerOptions {
readonly childCommand: string | null;
readonly processIds: ReadonlyArray<number>;
}>;
processTable?: Effect.Effect<
ReadonlyArray<{ readonly pid: number; readonly ppid: number; readonly name: string }>,
never
>;
subprocessPollIntervalMs?: number;
processKillGraceMs?: number;
maxRetainedInactiveSessions?: number;
Expand Down Expand Up @@ -248,6 +253,7 @@ const createManager = (
...(options.subprocessInspector !== undefined
? { subprocessInspector: options.subprocessInspector }
: {}),
...(options.processTable !== undefined ? { processTable: options.processTable } : {}),
...(options.subprocessPollIntervalMs !== undefined
? { subprocessPollIntervalMs: options.subprocessPollIntervalMs }
: {}),
Expand Down Expand Up @@ -1073,6 +1079,96 @@ it.layer(
}),
);

it("calculates snapshot failure backoff and success reset delays", () => {
assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 0), 1_000);
assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 1), 2_000);
assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 2), 4_000);
assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 30), 60_000);
});

it.effect("uses process snapshots from the resource monitor", () =>
Effect.gen(function* () {
let snapshotCalls = 0;
const { manager, getEvents } = yield* createManager(5, {
subprocessPollIntervalMs: 20,
processTable: Effect.sync(() => {
snapshotCalls += 1;
return [{ pid: 100, ppid: 9000, name: "ping.exe" }];
}),
}).pipe(Effect.provide(withHostPlatform("win32")));

yield* manager.open(openInput());
yield* waitFor(
Effect.map(getEvents, (events) =>
events.some(
(event) =>
event.type === "activity" && event.hasRunningSubprocess && event.label === "ping",
),
),
"1200 millis",
);
expect(snapshotCalls).toBeGreaterThan(0);
}),
);

it.effect("backs off the spawned fallback when the resource monitor snapshot fails", () =>
Effect.gen(function* () {
const fallbackCalls: Array<number> = [];
const processRunner: ProcessRunner.ProcessRunner["Service"] = {
run: () =>
Clock.currentTimeMillis.pipe(
Effect.map((now) => {
fallbackCalls.push(now);
return {
stdout: " 100 9000 vim",
stderr: "",
code: ChildProcessSpawner.ExitCode(0),
timedOut: false,
stdoutTruncated: false,
stderrInvalidUtf8: false,
stdoutInvalidUtf8: false,
stderrTruncated: false,
};
}),
),
};

const { manager, getEvents } = yield* createManager(5, {
subprocessPollIntervalMs: 20,
processTable: Effect.fail("sidecar unavailable").pipe(
Effect.mapError((cause) => cause as never),
),
}).pipe(
Effect.provideService(ProcessRunner.ProcessRunner, processRunner),
Effect.provide(withHostPlatform("linux")),
);

yield* manager.open(openInput());
// The fallback data is still applied while the sidecar is down.
yield* waitFor(
Effect.map(getEvents, (events) =>
events.some(
(event) =>
event.type === "activity" &&
event.hasRunningSubprocess === true &&
event.label === "vim",
),
),
"1200 millis",
);

yield* waitFor(
Effect.sync(() => fallbackCalls.length >= 4),
"2000 millis",
);
// Four snapshots at the 20 ms base cadence would span ~60 ms. Backoff
// (40 + 80 + 160 ms) stretches the same four snapshots past 150 ms, so
// a stalled sidecar no longer hot-loops the spawned fallback.
const spanMs = fallbackCalls[3]! - fallbackCalls[0]!;
expect(spanMs).toBeGreaterThan(150);
}),
);

it.effect("caps persisted history to configured line limit", () =>
Effect.gen(function* () {
const { manager, ptyAdapter } = yield* createManager(3);
Expand Down
Loading
Loading