diff --git a/apps/server/src/diagnostics/ProcessDiagnostics.test.ts b/apps/server/src/diagnostics/ProcessDiagnostics.test.ts index 2efa3375d275..5bcc74206893 100644 --- a/apps/server/src/diagnostics/ProcessDiagnostics.test.ts +++ b/apps/server/src/diagnostics/ProcessDiagnostics.test.ts @@ -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")), diff --git a/apps/server/src/resourceTelemetry/Model.test.ts b/apps/server/src/resourceTelemetry/Model.test.ts index 94690e3967bc..6f759ac9744f 100644 --- a/apps/server/src/resourceTelemetry/Model.test.ts +++ b/apps/server/src/resourceTelemetry/Model.test.ts @@ -39,7 +39,7 @@ function nativeSnapshot( sequence = 1, ): ResourceMonitorSnapshotEvent { return { - version: 2, + version: 3, type: "snapshot", sequence, sampledAtUnixMs, diff --git a/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts index 8a595bc8b480..8732f09e278b 100644 --- a/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts +++ b/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts @@ -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, @@ -92,6 +92,10 @@ 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.", @@ -99,8 +103,12 @@ describe("NativeTelemetryRequestTimedOut", () => { 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); }); }); diff --git a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts index 232079d9dc9b..0299e813bb55 100644 --- a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts +++ b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts @@ -5,6 +5,7 @@ import type { ResourceMonitorEvent, ResourceMonitorExternalProcess, ResourceMonitorHelloEvent, + ResourceMonitorProcessTableEntry, ResourceMonitorSnapshotEvent, ResourceTelemetrySourceStatus, } from "@t3tools/contracts"; @@ -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); @@ -76,7 +78,7 @@ export class NativeTelemetryHandshakeTimedOut extends Schema.TaggedErrorClass()( "NativeTelemetryRequestTimedOut", { - operation: Schema.Literals(["readHistory", "sampleNow"]), + operation: Schema.Literals(["processTable", "readHistory", "sampleNow"]), timeoutMs: Schema.Number, }, ) { @@ -192,6 +194,10 @@ export class NativeTelemetryClient extends Context.Service< snapshot: HostPowerSnapshot, ) => Effect.Effect; readonly sampleNow: Effect.Effect; + readonly processTable: Effect.Effect< + ReadonlyArray, + NativeTelemetryClientError + >; readonly retry: Effect.Effect; readonly health: Effect.Effect; readonly subscribeHealth: Effect.Effect< @@ -386,6 +392,12 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu const pendingSamples = yield* Ref.make( new Map>(), ); + const pendingProcessTables = yield* Ref.make( + new Map< + string, + Deferred.Deferred, NativeTelemetryClientError> + >(), + ); const pendingHistories = yield* Ref.make(new Map()); const snapshots = yield* PubSub.sliding(8); const healthChanges = yield* PubSub.sliding(4); @@ -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), @@ -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); @@ -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, + 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({ @@ -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)) @@ -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: diff --git a/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts b/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts index a96423607baf..9c371078332d 100644 --- a/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts +++ b/apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts @@ -82,7 +82,7 @@ function nativeSnapshot(input: { }), ]; return { - version: 2, + version: 3, type: "snapshot", sequence: input.sequence, sampledAtUnixMs: input.sampledAtUnixMs, @@ -497,7 +497,7 @@ describe("ResourceTelemetry", () => { const nativeHealth = yield* Ref.make({ status: "healthy", hello: Option.some({ - version: 2, + version: 3, type: "hello", sidecarVersion: "0.1.0", sidecarPid: 9_000, diff --git a/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.test.ts b/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.test.ts index 879c83d86dee..fff4588c8468 100644 --- a/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.test.ts +++ b/apps/server/src/resourceTelemetry/ResourceTelemetryHistory.test.ts @@ -64,7 +64,7 @@ function snapshot( }), ]; return { - version: 2, + version: 3, type: "snapshot", sequence, sampledAtUnixMs, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 3a93adc6d761..1578e7916e87 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -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( diff --git a/apps/server/src/terminal/Manager.test.ts b/apps/server/src/terminal/Manager.test.ts index 47d91e4516ec..0e6a4995bd95 100644 --- a/apps/server/src/terminal/Manager.test.ts +++ b/apps/server/src/terminal/Manager.test.ts @@ -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"; @@ -210,6 +211,10 @@ interface CreateManagerOptions { readonly childCommand: string | null; readonly processIds: ReadonlyArray; }>; + processTable?: Effect.Effect< + ReadonlyArray<{ readonly pid: number; readonly ppid: number; readonly name: string }>, + never + >; subprocessPollIntervalMs?: number; processKillGraceMs?: number; maxRetainedInactiveSessions?: number; @@ -248,6 +253,7 @@ const createManager = ( ...(options.subprocessInspector !== undefined ? { subprocessInspector: options.subprocessInspector } : {}), + ...(options.processTable !== undefined ? { processTable: options.processTable } : {}), ...(options.subprocessPollIntervalMs !== undefined ? { subprocessPollIntervalMs: options.subprocessPollIntervalMs } : {}), @@ -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 = []; + 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); diff --git a/apps/server/src/terminal/Manager.ts b/apps/server/src/terminal/Manager.ts index 64c2dbb913fb..a7c84e7793f1 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -26,6 +26,7 @@ import { type TerminalMetadataStreamEvent, type TerminalOpenInput, type TerminalResizeInput, + type ResourceMonitorProcessTableEntry, type TerminalRestartInput, type TerminalSessionSnapshot, type TerminalSessionStatus, @@ -59,6 +60,7 @@ import { } from "../observability/Metrics.ts"; import * as ProcessRunner from "../processRunner.ts"; import * as PortScanner from "../preview/PortScanner.ts"; +import * as NativeTelemetryClient from "../resourceTelemetry/NativeTelemetryClient.ts"; import * as PtyAdapter from "./PtyAdapter.ts"; export { @@ -77,6 +79,7 @@ export { const DEFAULT_HISTORY_LINE_LIMIT = 5_000; const DEFAULT_PERSIST_DEBOUNCE_MS = 40; const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000; +const MAX_SUBPROCESS_POLL_INTERVAL_MS = 60_000; const DEFAULT_PROCESS_KILL_GRACE_MS = 1_000; const DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS = 128; const DEFAULT_OPEN_COLS = 120; @@ -89,7 +92,7 @@ class TerminalSubprocessCheckError extends Schema.TaggedErrorClass; } +export function subprocessSnapshotPollDelayMs( + pollIntervalMs: number, + failureCount: number, +): number { + return Math.min(pollIntervalMs * 2 ** failureCount, MAX_SUBPROCESS_POLL_INTERVAL_MS); +} + function parsePosixProcessTable(stdout: string): TerminalProcessTableSnapshot { const childrenByParent = new Map(); const commandById = new Map(); @@ -643,15 +653,15 @@ function parsePosixProcessTable(stdout: string): TerminalProcessTableSnapshot { return { childrenByParent, commandById }; } -function parseWindowsProcessTable(stdout: string): TerminalProcessTableSnapshot { +function processTableSnapshotFromProcesses( + processes: ReadonlyArray, +): TerminalProcessTableSnapshot { const childrenByParent = new Map(); const commandById = new Map(); - for (const line of stdout.split(/\r?\n/g)) { - const [pidRaw, parentPidRaw, nameRaw] = line.trim().split("|", 3); - const pid = Number(pidRaw); - const parentPid = Number(parentPidRaw); + for (const process of processes) { + const { pid, ppid: parentPid, name } = process; if (!Number.isInteger(pid) || !Number.isInteger(parentPid)) continue; - commandById.set(pid, nameRaw?.trim() ?? ""); + commandById.set(pid, name.trim()); const children = childrenByParent.get(parentPid) ?? []; children.push(pid); childrenByParent.set(parentPid, children); @@ -746,14 +756,11 @@ const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnaps TerminalSubprocessCheckError, ProcessRunner.ProcessRunner > { + const processRunner = yield* ProcessRunner.ProcessRunner; const command = 'Get-CimInstance Win32_Process -ErrorAction Stop | ForEach-Object { Write-Output "$($_.ProcessId)|$($_.ParentProcessId)|$($_.Name)" }'; - const processRunner = yield* ProcessRunner.ProcessRunner; const result = yield* processRunner .run({ - // powershell.exe is a real executable — never spawn it through cmd.exe - // shell mode, which would re-tokenize the `-Command` payload (pipes, - // semicolons) before PowerShell ever sees it. command: "powershell.exe", args: ["-NoProfile", "-NonInteractive", "-Command", command], timeout: "1500 millis", @@ -763,16 +770,10 @@ const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnaps }) .pipe( Effect.mapError( - (cause) => - new TerminalSubprocessCheckError({ - cause, - command: "powershell", - }), + (cause) => new TerminalSubprocessCheckError({ cause, command: "powershell" }), ), ); if (result.code !== 0 || result.timedOut || result.stdoutTruncated) { - // Not authoritative: an empty or partial table would mark every terminal - // idle and clear its registered process ids. Failing skips the tick. return yield* new TerminalSubprocessCheckError({ command: "powershell", exitCode: result.code, @@ -780,7 +781,15 @@ const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnaps stdoutTruncated: result.stdoutTruncated, }); } - return parseWindowsProcessTable(result.stdout); + const processes = result.stdout.split(/\r?\n/g).flatMap((line) => { + const [pidRaw, ppidRaw, name = ""] = line.trim().split("|", 3); + const pid = Number(pidRaw); + const ppid = Number(ppidRaw); + return Number.isInteger(pid) && pid > 0 && Number.isInteger(ppid) + ? [{ pid, ppid, name }] + : []; + }); + return processTableSnapshotFromProcesses(processes); }, ); @@ -1115,6 +1124,10 @@ interface TerminalManagerOptions { shellResolver?: () => string; env?: NodeJS.ProcessEnv; subprocessInspector?: TerminalSubprocessInspector; + processTable?: Effect.Effect< + ReadonlyArray, + TerminalSubprocessCheckError + >; subprocessPollIntervalMs?: number; processKillGraceMs?: number; maxRetainedInactiveSessions?: number; @@ -1133,9 +1146,15 @@ export const make = Effect.fn("TerminalManager.make")(function* () { const { terminalLogsDir } = yield* ServerConfig.ServerConfig; const ptyAdapter = yield* PtyAdapter.PtyAdapter; const portDiscovery = yield* PortScanner.PortDiscovery; + const nativeTelemetry = yield* NativeTelemetryClient.NativeTelemetryClient; return yield* makeWithOptions({ logsDir: terminalLogsDir, ptyAdapter, + processTable: nativeTelemetry.processTable.pipe( + Effect.mapError( + (cause) => new TerminalSubprocessCheckError({ cause, command: "resource-monitor" }), + ), + ), registerTerminalProcesses: portDiscovery.registerTerminalProcesses, unregisterTerminal: portDiscovery.unregisterTerminal, }); @@ -1162,23 +1181,60 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func // One process-table snapshot per poll tick, shared across every terminal. // Per-terminal `pgrep`/`ps` calls multiply spawn load by terminal count and // can exhaust the PID space on hosts with many sessions (#6332). - const fetchProcessTableSnapshot = ( + const fallbackProcessTableSnapshot = ( platform === "win32" ? windowsProcessTableSnapshot() : posixProcessTableSnapshot(yield* resolvePosixPsCommand()) ).pipe(Effect.provideService(ProcessRunner.ProcessRunner, processRunner)); + const fetchProcessTableSnapshot: Effect.Effect< + { + readonly snapshot: TerminalProcessTableSnapshot; + /** + * False when the sidecar snapshot failed and this table came from the + * spawned fallback. The data is still applied, but the tick counts as + * a failure so polling backs off instead of hot-looping the fallback. + */ + readonly snapshotSucceeded: boolean; + }, + TerminalSubprocessCheckError + > = options.processTable + ? options.processTable.pipe( + Effect.map((entries) => ({ + snapshot: processTableSnapshotFromProcesses(entries), + snapshotSucceeded: true, + })), + Effect.catch(() => + fallbackProcessTableSnapshot.pipe( + Effect.map((snapshot) => ({ snapshot, snapshotSucceeded: false })), + ), + ), + ) + : fallbackProcessTableSnapshot.pipe( + Effect.map((snapshot) => ({ snapshot, snapshotSucceeded: true })), + ); const customSubprocessInspector = options.subprocessInspector; const acquireSubprocessInspector: Effect.Effect< - TerminalSubprocessInspector, + { + readonly inspector: TerminalSubprocessInspector; + readonly snapshotSucceeded: boolean; + }, TerminalSubprocessCheckError > = customSubprocessInspector !== undefined - ? Effect.succeed(customSubprocessInspector) + ? Effect.succeed({ inspector: customSubprocessInspector, snapshotSucceeded: true }) : Effect.map( fetchProcessTableSnapshot, - (snapshot): TerminalSubprocessInspector => - (terminalPid) => + ({ + snapshot, + snapshotSucceeded, + }): { + readonly inspector: TerminalSubprocessInspector; + readonly snapshotSucceeded: boolean; + } => ({ + inspector: (terminalPid) => Effect.succeed(deriveSubprocessInspectResult(snapshot, terminalPid, platform)), + snapshotSucceeded, + }), ); const subprocessPollIntervalMs = options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS; @@ -2008,7 +2064,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func ); if (runningSessions.length === 0) { - return; + return true; } const inspectorOption = yield* acquireSubprocessInspector.pipe( @@ -2016,15 +2072,22 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func Effect.catch((reason) => Effect.logWarning("failed to snapshot processes for terminal subprocess polling", { reason, - }).pipe(Effect.as(Option.none())), + }).pipe( + Effect.as( + Option.none<{ + readonly inspector: TerminalSubprocessInspector; + readonly snapshotSucceeded: boolean; + }>(), + ), + ), ), ); if (Option.isNone(inspectorOption)) { - return; + return false; } - const subprocessInspector = inspectorOption.value; + const { inspector: subprocessInspector, snapshotSucceeded } = inspectorOption.value; const checkSubprocessActivity = Effect.fn("terminal.checkSubprocessActivity")(function* ( session: TerminalSessionState & { pid: number }, @@ -2093,6 +2156,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func concurrency: "unbounded", discard: true, }); + return snapshotSucceeded; }); const hasRunningSessions = readManagerState.pipe( @@ -2101,14 +2165,26 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func ), ); + let subprocessSnapshotFailureCount = 0; yield* Effect.forever( hasRunningSessions.pipe( Effect.flatMap((active) => active ? pollSubprocessActivity().pipe( - Effect.flatMap(() => Effect.sleep(subprocessPollIntervalMs)), + Effect.flatMap((snapshotSucceeded) => { + subprocessSnapshotFailureCount = snapshotSucceeded + ? 0 + : Math.min(subprocessSnapshotFailureCount + 1, 30); + const delayMs = subprocessSnapshotPollDelayMs( + subprocessPollIntervalMs, + subprocessSnapshotFailureCount, + ); + return Effect.sleep(delayMs); + }), ) - : Effect.sleep(subprocessPollIntervalMs), + : Effect.sync(() => { + subprocessSnapshotFailureCount = 0; + }).pipe(Effect.flatMap(() => Effect.sleep(subprocessPollIntervalMs))), ), ), ).pipe(Effect.forkIn(workerScope)); diff --git a/docs/internals/resource-telemetry.md b/docs/internals/resource-telemetry.md index 0f3e6674ff79..cbc09ea14710 100644 --- a/docs/internals/resource-telemetry.md +++ b/docs/internals/resource-telemetry.md @@ -80,10 +80,12 @@ line on stdout: - `setSampleInterval` - `setStreaming` - `sampleNow` +- `processTable` - `readHistory` - `shutdown` - `hello` - `snapshot` +- `processTable` - `historyChunk` - `error` diff --git a/native/resource-monitor/src/main.rs b/native/resource-monitor/src/main.rs index 0596aea977cd..14975402121a 100644 --- a/native/resource-monitor/src/main.rs +++ b/native/resource-monitor/src/main.rs @@ -8,7 +8,7 @@ use sysinfo::{ MINIMUM_CPU_UPDATE_INTERVAL, Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind, }; -const PROTOCOL_VERSION: u32 = 2; +const PROTOCOL_VERSION: u32 = 3; const MIN_SAMPLE_INTERVAL_MS: u64 = 250; const MAX_SAMPLE_INTERVAL_MS: u64 = 60_000; const PROCESS_START_TIME_PRECISION_MS: u64 = 1_000; @@ -66,6 +66,10 @@ enum Command { version: u32, request_id: String, }, + ProcessTable { + version: u32, + request_id: String, + }, ReadHistory { version: u32, request_id: String, @@ -84,6 +88,7 @@ impl Command { | Self::SetSampleInterval { version, .. } | Self::SetStreaming { version, .. } | Self::SampleNow { version, .. } + | Self::ProcessTable { version, .. } | Self::ReadHistory { version, .. } | Self::Shutdown { version } => *version, } @@ -146,6 +151,24 @@ struct ProcessSample { io_semantics: IoSemantics, } +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ProcessTableEntry { + pid: u32, + ppid: u32, + name: String, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ProcessTableEvent<'a> { + version: u32, + #[serde(rename = "type")] + event_type: &'static str, + request_id: &'a str, + processes: Vec, +} + impl ProcessSample { fn estimated_history_bytes(&self) -> usize { std::mem::size_of::() @@ -351,6 +374,39 @@ impl Collector { self.cpu_baseline_refreshed_at = Some(Instant::now()); } + fn process_table(&mut self) -> Vec { + self.system.refresh_processes_specifics( + ProcessesToUpdate::All, + true, + ProcessRefreshKind::nothing().without_tasks(), + ); + let mut processes = self + .system + .processes() + .iter() + .filter_map(|(pid, process)| { + let pid = pid.as_u32(); + // Pid 0 is the kernel idle process on some platforms. The + // processTable contract requires positive pids, and one zero + // would fail the whole event decode on the server, so drop it + // here. It can never be a terminal descendant. + if pid == 0 { + return None; + } + Some(ProcessTableEntry { + pid, + ppid: process.parent().map(Pid::as_u32).unwrap_or(0), + name: truncate_utf8( + process.name().to_string_lossy().into_owned(), + MAX_PROCESS_NAME_BYTES, + ), + }) + }) + .collect::>(); + processes.sort_by_key(|process| process.pid); + processes + } + fn sample(&mut self, config: &CollectorConfig, request_id: Option) -> SnapshotEvent { if let Some(delay) = remaining_cpu_measurement_delay(self.cpu_baseline_refreshed_at.take(), Instant::now()) @@ -815,6 +871,15 @@ fn main() -> io::Result<()> { )?; } } + Command::ProcessTable { request_id, .. } => { + let event = ProcessTableEvent { + version: PROTOCOL_VERSION, + event_type: "processTable", + request_id: &request_id, + processes: collector.process_table(), + }; + write_event(&mut writer, &event)?; + } Command::ReadHistory { request_id, window_ms, @@ -892,7 +957,7 @@ mod tests { #[test] fn decodes_protocol_commands() { let configure = serde_json::from_str::( - r#"{"version":2,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#, + r#"{"version":3,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#, ) .expect("configure command"); @@ -912,7 +977,7 @@ mod tests { } let read_history = serde_json::from_str::( - r#"{"version":2,"type":"readHistory","requestId":"history-1","windowMs":60000}"#, + r#"{"version":3,"type":"readHistory","requestId":"history-1","windowMs":60000}"#, ) .expect("read history command"); assert!(matches!( @@ -923,6 +988,15 @@ mod tests { .. } if request_id == "history-1" )); + + let process_table = serde_json::from_str::( + r#"{"version":3,"type":"processTable","requestId":"processes-1"}"#, + ) + .expect("process table command"); + assert!(matches!( + process_table, + Command::ProcessTable { request_id, .. } if request_id == "processes-1" + )); } #[test] diff --git a/packages/contracts/src/resourceTelemetry.ts b/packages/contracts/src/resourceTelemetry.ts index 3ec1e4de3ef4..fc519c3bc64f 100644 --- a/packages/contracts/src/resourceTelemetry.ts +++ b/packages/contracts/src/resourceTelemetry.ts @@ -4,7 +4,7 @@ import { NonNegativeInt, PositiveInt, TrimmedNonEmptyString } from "./baseSchema import { HostPowerSnapshot } from "./background.ts"; import { DesktopUpdateStateSchema } from "./ipc.ts"; -export const RESOURCE_MONITOR_PROTOCOL_VERSION = 2 as const; +export const RESOURCE_MONITOR_PROTOCOL_VERSION = 3 as const; export const ResourceTelemetryIoSemantics = Schema.Literals([ "storage", @@ -102,6 +102,13 @@ export const ResourceMonitorSampleNowCommand = Schema.Struct({ }); export type ResourceMonitorSampleNowCommand = typeof ResourceMonitorSampleNowCommand.Type; +export const ResourceMonitorProcessTableCommand = Schema.Struct({ + version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), + type: Schema.Literal("processTable"), + requestId: TrimmedNonEmptyString, +}); +export type ResourceMonitorProcessTableCommand = typeof ResourceMonitorProcessTableCommand.Type; + export const ResourceMonitorSetSampleIntervalCommand = Schema.Struct({ version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), type: Schema.Literal("setSampleInterval"), @@ -137,6 +144,7 @@ export const ResourceMonitorCommand = Schema.Union([ ResourceMonitorSetSampleIntervalCommand, ResourceMonitorSetStreamingCommand, ResourceMonitorSampleNowCommand, + ResourceMonitorProcessTableCommand, ResourceMonitorReadHistoryCommand, ResourceMonitorShutdownCommand, ]); @@ -168,6 +176,21 @@ export const ResourceMonitorSnapshotEvent = Schema.Struct({ }); export type ResourceMonitorSnapshotEvent = typeof ResourceMonitorSnapshotEvent.Type; +export const ResourceMonitorProcessTableEntry = Schema.Struct({ + pid: PositiveInt, + ppid: NonNegativeInt, + name: Schema.String, +}); +export type ResourceMonitorProcessTableEntry = typeof ResourceMonitorProcessTableEntry.Type; + +export const ResourceMonitorProcessTableEvent = Schema.Struct({ + version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), + type: Schema.Literal("processTable"), + requestId: TrimmedNonEmptyString, + processes: Schema.Array(ResourceMonitorProcessTableEntry), +}); +export type ResourceMonitorProcessTableEvent = typeof ResourceMonitorProcessTableEvent.Type; + export const ResourceMonitorHistoryChunkEvent = Schema.Struct({ version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), type: Schema.Literal("historyChunk"), @@ -189,6 +212,7 @@ export type ResourceMonitorErrorEvent = typeof ResourceMonitorErrorEvent.Type; export const ResourceMonitorEvent = Schema.Union([ ResourceMonitorHelloEvent, ResourceMonitorSnapshotEvent, + ResourceMonitorProcessTableEvent, ResourceMonitorHistoryChunkEvent, ResourceMonitorErrorEvent, ]);