From ea646c0834a3394ecb0be4a30c5d367e5a9002bd Mon Sep 17 00:00:00 2001 From: Utkarsh Patil <73941998+UtkarshUsername@users.noreply.github.com> Date: Mon, 7 Sep 2026 01:17:38 +0530 Subject: [PATCH] fix(server): stop Windows terminal polling from spiking CPU (#9476) --- .../diagnostics/ProcessDiagnostics.test.ts | 2 +- .../src/resourceTelemetry/Model.test.ts | 2 +- .../NativeTelemetryClient.ts | 93 +++++++++++- .../ResourceTelemetry.test.ts | 4 +- .../ResourceTelemetryHistory.test.ts | 2 +- apps/server/src/server.ts | 1 + apps/server/src/terminal/Manager.test.ts | 96 +++++++++++++ apps/server/src/terminal/Manager.ts | 136 ++++++++++++++---- native/resource-monitor/src/main.rs | 82 ++++++++++- packages/contracts/src/resourceTelemetry.ts | 26 +++- 10 files changed, 404 insertions(+), 40 deletions(-) diff --git a/apps/server/src/diagnostics/ProcessDiagnostics.test.ts b/apps/server/src/diagnostics/ProcessDiagnostics.test.ts index 2efa3375d275d8e177a56c03890fb2731804a58c..5bcc742068935b0de0eca1378ee3ee5b3eaf39b2 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 94690e3967bc49366356018bcf6b83b614a7e983..6f759ac9744fdfdf56676504bbd55f40f8a002d1 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.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts index 4af8f1b762d540885023ca4f48721be11153e4ec..9ddc72d61ebdb6891ebb1d916f66f2a0938e2fff 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< @@ -382,6 +388,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); @@ -399,10 +411,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), @@ -481,6 +497,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); @@ -936,6 +967,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({ @@ -957,6 +1042,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)) @@ -1010,6 +1096,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 a96423607bafae3b64d30aa0ac55a0edb2ea2d1c..9c371078332dc940293176dfb8cda604cb3db807 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 879c83d86deeba9eece6d57d7048c5db756166e2..fff4588c8468e0596c70927f60a061218a78c35f 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 696210b36b86a3f55655f42cad9d73dde209af91..349644966f26e810c2494413a704d357983b252e 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -374,6 +374,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 e480e11588b0c7fd72c31334e0f83ba5f4acfcc9..f631992e7ae3d1d672e1f38d52831176f0b022ff 100644 --- a/apps/server/src/terminal/Manager.test.ts +++ b/apps/server/src/terminal/Manager.test.ts @@ -14,6 +14,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 Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -220,6 +221,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; @@ -265,6 +270,7 @@ const createManager = ( ...(options.subprocessInspector !== undefined ? { subprocessInspector: options.subprocessInspector } : {}), + ...(options.processTable !== undefined ? { processTable: options.processTable } : {}), ...(options.subprocessPollIntervalMs !== undefined ? { subprocessPollIntervalMs: options.subprocessPollIntervalMs } : {}), @@ -1192,6 +1198,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 d9bdc6bcd92a93855c8508370eacba20f0a2cbd8..10143b7be20b8b9c91e5cab49fa034746e80519a 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -28,6 +28,7 @@ import { type TerminalMetadataStreamEvent, type TerminalOpenInput, type TerminalResizeInput, + type ResourceMonitorProcessTableEntry, type TerminalRestartInput, type TerminalSessionSnapshot, type TerminalSessionStatus, @@ -70,6 +71,7 @@ import { import { expandHomePath } from "../pathExpansion.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 { @@ -92,6 +94,7 @@ const DEFAULT_HISTORY_BYTE_LIMIT = 8 * 1024 * 1024; const MAX_HISTORY_CHUNK_LENGTH = 16 * 1024; 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; @@ -106,7 +109,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(); @@ -660,15 +670,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); @@ -763,14 +773,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", @@ -780,16 +787,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, @@ -797,7 +798,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); }, ); @@ -1310,6 +1319,10 @@ interface TerminalManagerOptions { shellResolver?: () => string; env?: NodeJS.ProcessEnv; subprocessInspector?: TerminalSubprocessInspector; + processTable?: Effect.Effect< + ReadonlyArray, + TerminalSubprocessCheckError + >; subprocessPollIntervalMs?: number; processKillGraceMs?: number; maxRetainedInactiveSessions?: number; @@ -1376,6 +1389,7 @@ 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; const serverSettings = yield* ServerSettings.ServerSettingsService; const path = yield* Path.Path; const resolveProviderInstanceEnvironment = Effect.fn( @@ -1391,6 +1405,11 @@ export const make = Effect.fn("TerminalManager.make")(function* () { 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, resolveProviderInstanceEnvironment, @@ -1437,23 +1456,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; @@ -2305,7 +2361,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func ); if (runningSessions.length === 0) { - return; + return true; } const inspectorOption = yield* acquireSubprocessInspector.pipe( @@ -2313,15 +2369,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 }, @@ -2390,6 +2453,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func concurrency: "unbounded", discard: true, }); + return snapshotSucceeded; }); const hasRunningSessions = readManagerState.pipe( @@ -2398,14 +2462,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/native/resource-monitor/src/main.rs b/native/resource-monitor/src/main.rs index 78ba5a271cd3218531b531c9fbaf0ed2b1da067a..6137544ea0e2c94bbc0101c07cda1992fe63d02b 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,41 @@ impl Collector { self.cpu_baseline_refreshed_at = Some(Instant::now()); } + fn process_table(&self) -> Vec { + // Use a dedicated System so this refresh cannot reset the CPU + // baseline tracked by self.system for snapshots. + let mut process_table_system = System::new(); + process_table_system.refresh_processes_specifics( + ProcessesToUpdate::All, + true, + ProcessRefreshKind::nothing().without_tasks(), + ); + let mut processes = process_table_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()) @@ -871,6 +929,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, @@ -981,7 +1048,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"); @@ -1001,7 +1068,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!( @@ -1012,6 +1079,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 87b52993a56b7a7b091b6cac7f31eefb66a930b4..ee9d2b3ac2588f33965e7c15512e893bf1ac8dc9 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; /** Whole-host capacity, independent of T3's process diagnostics. */ export const HostResourcesSnapshot = Schema.Struct({ @@ -112,6 +112,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"), @@ -147,6 +154,7 @@ export const ResourceMonitorCommand = Schema.Union([ ResourceMonitorSetSampleIntervalCommand, ResourceMonitorSetStreamingCommand, ResourceMonitorSampleNowCommand, + ResourceMonitorProcessTableCommand, ResourceMonitorReadHistoryCommand, ResourceMonitorShutdownCommand, ]); @@ -178,6 +186,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"), @@ -199,6 +222,7 @@ export type ResourceMonitorErrorEvent = typeof ResourceMonitorErrorEvent.Type; export const ResourceMonitorEvent = Schema.Union([ ResourceMonitorHelloEvent, ResourceMonitorSnapshotEvent, + ResourceMonitorProcessTableEvent, ResourceMonitorHistoryChunkEvent, ResourceMonitorErrorEvent, ]); -- 2.54.0