diff --git a/apps/server/src/diagnostics/ProcessDiagnostics.test.ts b/apps/server/src/diagnostics/ProcessDiagnostics.test.ts index 5bcc74206893..4e9f676379f5 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: 3, + version: 4, type: "snapshot", sequence: 1, sampledAtUnixMs: DateTime.toEpochMillis(DateTime.makeUnsafe("2026-05-05T10:00:00.000Z")), diff --git a/apps/server/src/preview/PortScanner.test.ts b/apps/server/src/preview/PortScanner.test.ts index 7fa15defeca9..477663e06d0d 100644 --- a/apps/server/src/preview/PortScanner.test.ts +++ b/apps/server/src/preview/PortScanner.test.ts @@ -10,6 +10,7 @@ import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Net from "@t3tools/shared/Net"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -20,8 +21,10 @@ import * as Scope from "effect/Scope"; import * as TestClock from "effect/testing/TestClock"; import { expect } from "vite-plus/test"; import { FetchHttpClient } from "effect/unstable/http"; +import { ChildProcessSpawner } from "effect/unstable/process"; import * as ProcessRunner from "../processRunner.ts"; +import * as NativeTelemetryClient from "../resourceTelemetry/NativeTelemetryClient.ts"; import * as PortScanner from "./PortScanner.ts"; const processProbeFailure: ProcessRunner.ProcessRunner["Service"]["run"] = (input) => Effect.fail( @@ -41,6 +44,7 @@ const processProbeFailure: ProcessRunner.ProcessRunner["Service"]["run"] = (inpu const TestProcessRunner = Layer.succeed(ProcessRunner.ProcessRunner, { run: processProbeFailure, }); +const TestNativeTelemetry = NativeTelemetryClient.layerTest(); let integrationListeningPort: number | null = null; @@ -68,6 +72,7 @@ const makeProbeFailureLayer = ( findAvailablePort: (preferred) => Effect.succeed(preferred), }), Layer.succeed(HostProcessPlatform, "linux"), + TestNativeTelemetry, FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetch))), ), ), @@ -79,6 +84,7 @@ const TestPortDiscoveryLive = PortScanner.layer.pipe( TestProcessRunner, TestIntegrationNet, Layer.succeed(HostProcessPlatform, "win32"), + TestNativeTelemetry, FetchHttpClient.layer, ), ), @@ -114,6 +120,7 @@ const makeLsofScannerLayer = (input: { findAvailablePort: (preferred) => Effect.succeed(preferred), }), Layer.succeed(HostProcessPlatform, "linux"), + TestNativeTelemetry, FetchHttpClient.layer.pipe( Layer.provide(Layer.succeed(FetchHttpClient.Fetch, input.fetch)), ), @@ -121,6 +128,31 @@ const makeLsofScannerLayer = (input: { ), ); +const makeWindowsScannerLayer = (input: { + readonly windowsListeners: NativeTelemetryClient.NativeTelemetryClient["Service"]["windowsListeners"]; + readonly run: ProcessRunner.ProcessRunner["Service"]["run"]; + readonly fetch?: typeof globalThis.fetch; +}) => + PortScanner.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(ProcessRunner.ProcessRunner, { run: input.run }), + Layer.succeed(Net.NetService, { + canListenOnHost: () => Effect.succeed(true), + isPortAvailableOnLoopback: () => Effect.succeed(true), + hasListenerOnHost: () => Effect.succeed(false), + reserveLoopbackPort: () => Effect.succeed(40_000), + findAvailablePort: (preferred) => Effect.succeed(preferred), + }), + Layer.succeed(HostProcessPlatform, "win32"), + NativeTelemetryClient.layerTest({ windowsListeners: input.windowsListeners }), + FetchHttpClient.layer.pipe( + Layer.provide(Layer.succeed(FetchHttpClient.Fetch, input.fetch ?? globalThis.fetch)), + ), + ), + ), + ); + const openServer = ( port: number, onConnection: (socket: NodeNet.Socket) => void, @@ -244,6 +276,337 @@ effectIt.layer(TestPortDiscoveryLive)("PortDiscovery integration (TCP probe fall ); }); +effectIt.effect("uses native Windows listeners without spawning PowerShell", () => { + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.succeed([{ port: LSOF_TEST_PORT, pid: 4_242, processName: "node" }]), + run: (input) => { + fallbackRuns += 1; + return processProbeFailure(input); + }, + fetch: ((_input: Parameters[0]) => + Promise.resolve( + new Response("app", { headers: { "content-type": "text/html" } }), + )) as typeof globalThis.fetch, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + yield* scanner.registerTerminalProcesses({ + threadId: "thread-1", + terminalId: "default", + processIds: [4_242], + }); + const servers = yield* scanner.scan(); + + expect(fallbackRuns).toBe(0); + expect(servers).toEqual([ + { + host: "localhost", + port: LSOF_TEST_PORT, + url: `http://localhost:${LSOF_TEST_PORT}`, + processName: "node", + pid: 4_242, + terminal: { threadId: "thread-1", terminalId: "default" }, + }, + ]); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("keeps a native Windows snapshot when both discovery paths later fail", () => { + let nativeCalls = 0; + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.suspend(() => { + nativeCalls += 1; + return nativeCalls === 1 + ? Effect.succeed([{ port: 43_123, pid: 4_242, processName: "node" }]) + : Effect.fail(new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" })); + }), + run: (input) => { + fallbackRuns += 1; + return processProbeFailure(input); + }, + fetch: ((_input: Parameters[0]) => + Promise.resolve( + new Response("app", { headers: { "content-type": "text/html" } }), + )) as typeof globalThis.fetch, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + expect((yield* scanner.scan())[0]?.port).toBe(43_123); + + yield* scanner.registerTerminalProcesses({ + threadId: "thread-1", + terminalId: "default", + processIds: [4_242], + }); + const retained = yield* scanner.scan(); + + expect(retained[0]?.port).toBe(43_123); + expect(retained[0]?.terminal).toEqual({ threadId: "thread-1", terminalId: "default" }); + expect(fallbackRuns).toBe(1); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("backs off every failed Windows PowerShell fallback", () => { + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.fail( + new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" }), + ), + run: (input) => { + fallbackRuns += 1; + return processProbeFailure(input); + }, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + yield* scanner.scan(); + yield* scanner.scan(); + expect(fallbackRuns).toBe(1); + + yield* TestClock.adjust(Duration.seconds(3)); + yield* scanner.scan(); + expect(fallbackRuns).toBe(2); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("starts the Windows fallback cooldown after a slow failure completes", () => { + let fallbackRuns = 0; + let fallbackStarted: Deferred.Deferred | null = null; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.fail( + new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" }), + ), + run: (input) => + Effect.sync(() => { + fallbackRuns += 1; + }).pipe( + Effect.andThen( + fallbackStarted === null ? Effect.void : Deferred.succeed(fallbackStarted, undefined), + ), + Effect.andThen(Effect.sleep(Duration.seconds(15))), + Effect.andThen(processProbeFailure(input)), + ), + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + fallbackStarted = yield* Deferred.make(); + const firstScan = yield* scanner.scan().pipe(Effect.forkChild); + yield* Deferred.await(fallbackStarted); + yield* TestClock.adjust(Duration.seconds(15)); + yield* Fiber.join(firstScan); + + yield* scanner.scan(); + expect(fallbackRuns).toBe(1); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("keeps the cached Windows snapshot after a nonzero fallback exit", () => { + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.fail( + new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" }), + ), + run: () => { + fallbackRuns += 1; + return Effect.succeed({ + stdout: fallbackRuns === 1 ? `127.0.0.1|${LSOF_TEST_PORT}|4242|node\n` : "", + stderr: "", + code: ChildProcessSpawner.ExitCode(fallbackRuns === 1 ? 0 : 1), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }); + }, + fetch: ((_input: Parameters[0]) => + Promise.resolve( + new Response("app", { headers: { "content-type": "text/html" } }), + )) as typeof globalThis.fetch, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + expect(yield* scanner.scan()).toHaveLength(1); + yield* TestClock.adjust(Duration.seconds(3)); + expect(yield* scanner.scan()).toHaveLength(1); + expect(yield* scanner.scan()).toHaveLength(1); + expect(fallbackRuns).toBe(2); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("keeps the cached Windows snapshot after truncated fallback output", () => { + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.fail( + new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" }), + ), + run: () => { + fallbackRuns += 1; + return Effect.succeed({ + stdout: `127.0.0.1|${LSOF_TEST_PORT}|4242|node\n`, + stderr: "", + code: ChildProcessSpawner.ExitCode(0), + timedOut: false, + stdoutTruncated: fallbackRuns > 1, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }); + }, + fetch: ((_input: Parameters[0]) => + Promise.resolve( + new Response("app", { headers: { "content-type": "text/html" } }), + )) as typeof globalThis.fetch, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + expect(yield* scanner.scan()).toHaveLength(1); + yield* TestClock.adjust(Duration.seconds(3)); + expect(yield* scanner.scan()).toHaveLength(1); + expect(yield* scanner.scan()).toHaveLength(1); + expect(fallbackRuns).toBe(2); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("keeps the last Windows fallback snapshot when a retry fails", () => { + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.fail( + new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" }), + ), + run: (input) => { + fallbackRuns += 1; + if (fallbackRuns > 1) return processProbeFailure(input); + return Effect.succeed({ + stdout: `127.0.0.1|${LSOF_TEST_PORT}|4242|node\n`, + stderr: "", + code: ChildProcessSpawner.ExitCode(0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }); + }, + fetch: ((_input: Parameters[0]) => + Promise.resolve( + new Response("app", { headers: { "content-type": "text/html" } }), + )) as typeof globalThis.fetch, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + expect(yield* scanner.scan()).toHaveLength(1); + + yield* TestClock.adjust(Duration.seconds(3)); + expect(yield* scanner.scan()).toHaveLength(1); + expect(yield* scanner.scan()).toHaveLength(1); + expect(fallbackRuns).toBe(2); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("refreshes terminal ownership when reusing a Windows fallback snapshot", () => { + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.fail( + new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" }), + ), + run: (input) => { + fallbackRuns += 1; + if (fallbackRuns > 1) return processProbeFailure(input); + return Effect.succeed({ + stdout: `127.0.0.1|${LSOF_TEST_PORT}|4242|node\n`, + stderr: "", + code: ChildProcessSpawner.ExitCode(0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }); + }, + fetch: ((_input: Parameters[0]) => + Promise.resolve( + new Response("app", { headers: { "content-type": "text/html" } }), + )) as typeof globalThis.fetch, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + yield* scanner.registerTerminalProcesses({ + threadId: "thread-1", + terminalId: "old", + processIds: [4_242], + }); + expect((yield* scanner.scan())[0]?.terminal).toEqual({ + threadId: "thread-1", + terminalId: "old", + }); + + yield* scanner.unregisterTerminal({ threadId: "thread-1", terminalId: "old" }); + yield* scanner.registerTerminalProcesses({ + threadId: "thread-2", + terminalId: "new", + processIds: [4_242], + }); + yield* TestClock.adjust(Duration.seconds(3)); + + expect((yield* scanner.scan())[0]?.terminal).toEqual({ + threadId: "thread-2", + terminalId: "new", + }); + expect(fallbackRuns).toBe(2); + }).pipe(Effect.provide(layer)); +}); + +effectIt.effect("runs a later Windows scan normally after an interrupted scan", () => { + let fallbackRuns = 0; + const layer = makeWindowsScannerLayer({ + windowsListeners: Effect.fail( + new NativeTelemetryClient.NativeTelemetryUnavailable({ reason: "test" }), + ), + run: () => { + fallbackRuns += 1; + return fallbackRuns === 1 + ? Effect.interrupt + : Effect.succeed({ + stdout: `127.0.0.1|${LSOF_TEST_PORT}|4242|node\n`, + stderr: "", + code: ChildProcessSpawner.ExitCode(0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }); + }, + fetch: ((_input: Parameters[0]) => + Promise.resolve( + new Response("app", { headers: { "content-type": "text/html" } }), + )) as typeof globalThis.fetch, + }); + + return Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + const interrupted = yield* scanner.scan().pipe(Effect.exit); + expect(Exit.isFailure(interrupted)).toBe(true); + if (Exit.isFailure(interrupted)) { + expect(Cause.hasInterruptsOnly(interrupted.cause)).toBe(true); + } + + expect(yield* scanner.scan()).toHaveLength(1); + expect(fallbackRuns).toBe(2); + }).pipe(Effect.provide(layer)); +}); + effectIt.effect("revalidates a successful HTML probe after its cache entry expires", () => { let responds = true; const requests: string[] = []; diff --git a/apps/server/src/preview/PortScanner.ts b/apps/server/src/preview/PortScanner.ts index f4d73d62320d..8f87c9d39f89 100644 --- a/apps/server/src/preview/PortScanner.ts +++ b/apps/server/src/preview/PortScanner.ts @@ -5,8 +5,11 @@ * stable line-prefixed field format; this is the only `lsof` flag set we rely * on). * - * Windows / lsof missing: checks a curated list of common dev ports through - * the shared Net service. + * Windows: asks the persistent resource monitor for the native TCP listener + * table. If the sidecar is unavailable, a backed-off PowerShell probe runs. + * + * lsof / Windows probe missing: checks a curated list of common dev ports + * through the shared Net service. * * Listening ports are published only after a bounded HTTP(S) probe finds a * successful HTML document or a redirect to one. @@ -21,6 +24,7 @@ import { PREVIEW_URL_MAX_LENGTH, ThreadId, type DiscoveredLocalServer, + type ResourceMonitorWindowsListener, } from "@t3tools/contracts"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Net from "@t3tools/shared/Net"; @@ -39,6 +43,7 @@ import * as Semaphore from "effect/Semaphore"; import { FetchHttpClient, HttpClient } from "effect/unstable/http"; import * as ProcessRunner from "../processRunner.ts"; +import * as NativeTelemetryClient from "../resourceTelemetry/NativeTelemetryClient.ts"; export class PortDiscovery extends Context.Service< PortDiscovery, @@ -72,7 +77,10 @@ export const COMMON_DEV_PORTS: ReadonlyArray = Object.freeze([ const POLL_INTERVAL = Duration.seconds(3); const LSOF_TIMEOUT_MS = 5_000; -const WINDOWS_LISTENER_TIMEOUT_MS = 5_000; +const WINDOWS_LISTENER_TIMEOUT_MS = 15_000; +const WINDOWS_FALLBACK_MAX_RETRY_MS = 60_000; +const WINDOWS_LISTENER_COMMAND = + '$m = @{}; Get-Process | ForEach-Object { $m[$_.Id] = $_.ProcessName }; Get-NetTCPConnection -State Listen -ErrorAction Stop | ForEach-Object { Write-Output "$($_.LocalAddress)|$($_.LocalPort)|$($_.OwningProcess)|$($m[[int]$_.OwningProcess])" }'; const WEB_PROBE_TIMEOUT = Duration.seconds(1); const WEB_PROBE_CACHE_TTL_MS = Duration.toMillis(Duration.seconds(15)); const WEB_PROBE_CONCURRENCY = 16; @@ -265,6 +273,45 @@ const parseWindowsListenerOutput = ( return [...seen.values()].toSorted((left, right) => left.port - right.port); }; +const windowsListenersToServers = ( + listeners: ReadonlyArray, + terminalByProcessId: ReadonlyMap = new Map(), +): ReadonlyArray => { + const seen = new Map(); + for (const listener of listeners) { + if (seen.has(listener.port)) continue; + seen.set(listener.port, { + host: "localhost", + port: listener.port, + url: `http://localhost:${listener.port}`, + processName: listener.processName?.trim() || null, + pid: listener.pid, + terminal: terminalByProcessId.get(listener.pid) ?? null, + }); + } + return [...seen.values()].toSorted((left, right) => left.port - right.port); +}; + +const withCurrentTerminalOwners = ( + servers: ReadonlyArray, + terminalByProcessId: ReadonlyMap, +): ReadonlyArray => + servers.map((server) => ({ + ...server, + terminal: server.pid === null ? null : (terminalByProcessId.get(server.pid) ?? null), + })); + +function windowsFallbackRetryDelayMs(failureCount: number): number { + return Math.min(3_000 * 2 ** Math.max(0, failureCount - 1), WINDOWS_FALLBACK_MAX_RETRY_MS); +} + +function windowsFallbackSuccessDelayMs(elapsedMs: number): number { + return Math.min( + Math.max(Duration.toMillis(POLL_INTERVAL), Math.max(0, elapsedMs) * 4), + WINDOWS_FALLBACK_MAX_RETRY_MS, + ); +} + const serversEqual = ( left: ReadonlyArray, right: ReadonlyArray, @@ -293,6 +340,7 @@ const serversEqual = ( export const make = Effect.gen(function* PortDiscoveryMake() { const net = yield* Net.NetService; const processRunner = yield* ProcessRunner.ProcessRunner; + const nativeTelemetry = yield* NativeTelemetryClient.NativeTelemetryClient; const hostPlatform = yield* HostProcessPlatform; const httpClient = (yield* HttpClient.HttpClient).pipe(HttpClient.withScope); const stateRef = yield* Ref.make({ @@ -302,6 +350,11 @@ export const make = Effect.gen(function* PortDiscoveryMake() { }); const webProbeCacheRef = yield* Ref.make>(new Map()); const scanSemaphore = yield* Semaphore.make(1); + const windowsFallbackRef = yield* Ref.make({ + failureCount: 0, + nextAttemptAtMillis: 0, + lastSnapshot: null as ReadonlyArray | null, + }); const probeCommonPorts = Effect.fn("PortDiscovery.probeCommonPorts")(function* () { const results = yield* Effect.forEach( @@ -478,6 +531,73 @@ export const make = Effect.gen(function* PortDiscoveryMake() { platform: hostPlatform, }).pipe(Effect.as(null)); + const probeWindowsFallback = Effect.fn("PortDiscovery.probeWindowsFallback")(function* ( + terminalByProcessId: ReadonlyMap, + ) { + const nowMillis = yield* Clock.currentTimeMillis; + const fallback = yield* Ref.get(windowsFallbackRef); + if (nowMillis < fallback.nextAttemptAtMillis) { + return fallback.lastSnapshot === null + ? yield* probeCommonPorts() + : withCurrentTerminalOwners(fallback.lastSnapshot, terminalByProcessId); + } + + const startedAtMillis = nowMillis; + const recoverWindowsProbeFailure = recoverProcessProbeFailure("windows-listeners"); + const listeners = yield* processRunner + .run({ + command: "powershell.exe", + args: ["-NoProfile", "-NonInteractive", "-Command", WINDOWS_LISTENER_COMMAND], + timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS), + maxOutputBytes: 1024 * 1024, + outputMode: "truncate", + }) + .pipe( + Effect.flatMap((result) => + result.code === 0 && !result.timedOut && !result.stdoutTruncated + ? Effect.succeed(parseWindowsListenerOutput(result.stdout, terminalByProcessId)) + : Effect.logDebug( + "preview port process probe returned an incomplete result; falling back to common-port probes", + { + probe: "windows-listeners", + platform: hostPlatform, + exitCode: result.code, + timedOut: result.timedOut, + stdoutTruncated: result.stdoutTruncated, + }, + ).pipe(Effect.as(null)), + ), + Effect.catchTags({ + ProcessSpawnError: recoverWindowsProbeFailure, + ProcessStdinError: recoverWindowsProbeFailure, + ProcessOutputLimitError: recoverWindowsProbeFailure, + ProcessReadError: recoverWindowsProbeFailure, + ProcessTimeoutError: recoverWindowsProbeFailure, + }), + ); + if (listeners !== null) { + const completedAtMillis = yield* Clock.currentTimeMillis; + yield* Ref.set(windowsFallbackRef, { + failureCount: 0, + nextAttemptAtMillis: + completedAtMillis + windowsFallbackSuccessDelayMs(completedAtMillis - startedAtMillis), + lastSnapshot: listeners, + }); + return listeners; + } + + const failureCount = fallback.failureCount + 1; + const completedAtMillis = yield* Clock.currentTimeMillis; + yield* Ref.set(windowsFallbackRef, { + failureCount, + nextAttemptAtMillis: completedAtMillis + windowsFallbackRetryDelayMs(failureCount), + lastSnapshot: fallback.lastSnapshot, + }); + return fallback.lastSnapshot === null + ? yield* probeCommonPorts() + : withCurrentTerminalOwners(fallback.lastSnapshot, terminalByProcessId); + }); + const scanUnlocked = Effect.fn("PortDiscovery.scanUnlocked")(function* ( configuredUrls: ReadonlyArray, ) { @@ -489,29 +609,23 @@ export const make = Effect.gen(function* PortDiscoveryMake() { } } if (hostPlatform === "win32") { - const recoverWindowsProbeFailure = recoverProcessProbeFailure("windows-listeners"); - const command = - 'Get-NetTCPConnection -State Listen -ErrorAction Stop | ForEach-Object { $processName = (Get-Process -Id $_.OwningProcess -ErrorAction SilentlyContinue).ProcessName; Write-Output "$($_.LocalAddress)|$($_.LocalPort)|$($_.OwningProcess)|$processName" }'; - const listeners = yield* processRunner - .run({ - command: "powershell.exe", - args: ["-NoProfile", "-NonInteractive", "-Command", command], - timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS), - maxOutputBytes: 1024 * 1024, - outputMode: "truncate", - }) - .pipe( - Effect.map((result) => parseWindowsListenerOutput(result.stdout, terminalByProcessId)), - Effect.catchTags({ - ProcessSpawnError: recoverWindowsProbeFailure, - ProcessStdinError: recoverWindowsProbeFailure, - ProcessOutputLimitError: recoverWindowsProbeFailure, - ProcessReadError: recoverWindowsProbeFailure, - ProcessTimeoutError: recoverWindowsProbeFailure, - }), - ); - if (listeners !== null) return yield* probeWebServers(listeners, configuredUrls); - return yield* probeWebServers(yield* probeCommonPorts(), configuredUrls); + const nativeListeners = yield* nativeTelemetry.windowsListeners.pipe( + Effect.map((listeners) => windowsListenersToServers(listeners, terminalByProcessId)), + Effect.catch((cause) => + Effect.logDebug("native Windows listener discovery failed; using fallback", { + cause, + }).pipe(Effect.as(null)), + ), + ); + const listeners = + nativeListeners === null + ? yield* probeWindowsFallback(terminalByProcessId) + : yield* Ref.set(windowsFallbackRef, { + failureCount: 0, + nextAttemptAtMillis: 0, + lastSnapshot: nativeListeners, + }).pipe(Effect.as(nativeListeners)); + return yield* probeWebServers(listeners, configuredUrls); } const recoverLsofProbeFailure = recoverProcessProbeFailure("lsof"); const lsofResult = yield* processRunner diff --git a/apps/server/src/resourceTelemetry/Model.test.ts b/apps/server/src/resourceTelemetry/Model.test.ts index 6f759ac9744f..6247215854d6 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: 3, + version: 4, type: "snapshot", sequence, sampledAtUnixMs, diff --git a/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts index 7d365d0c0bac..ec4e7019dfc4 100644 --- a/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts +++ b/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts @@ -1,20 +1,41 @@ -import type { HostPowerSnapshot } from "@t3tools/contracts"; +import { + RESOURCE_MONITOR_PROTOCOL_VERSION, + ResourceMonitorCommand as ResourceMonitorCommandSchema, + ResourceMonitorEvent as ResourceMonitorEventSchema, + type HostPowerSnapshot, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Semaphore from "effect/Semaphore"; +import * as Schema from "effect/Schema"; +import * as Sink from "effect/Sink"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { ChildProcessSpawner } from "effect/unstable/process"; +import * as ServerConfig from "../config.ts"; import { canCommandNativeTelemetrySidecar, canRequestNativeTelemetryRetry, commitCollectionControlUpdate, + layer as nativeTelemetryClientLayer, + NativeTelemetryExited, retainRecentNativeTelemetryFailures, resolveNativeSampleIntervalMs, + runPendingNativeTelemetryRequest, synchronizeCollectionControlOnStart, + NativeTelemetryClient, + type NativeTelemetryClientError, } from "./NativeTelemetryClient.ts"; +import * as ResourceMonitorBinary from "./ResourceMonitorBinary.ts"; const basePower: HostPowerSnapshot = { source: "electron-main", @@ -29,6 +50,11 @@ const basePower: HostPowerSnapshot = { updatedAt: DateTime.makeUnsafe("2026-06-17T12:00:00.000Z"), }; +const decodeMonitorCommand = Schema.decodeUnknownSync( + Schema.fromJsonString(ResourceMonitorCommandSchema), +); +const encodeMonitorEvent = Schema.encodeSync(Schema.fromJsonString(ResourceMonitorEventSchema)); + describe("resolveNativeSampleIntervalMs", () => { it("keeps a recovery cadence while suspended and backs off under host constraints", () => { expect(resolveNativeSampleIntervalMs({ ...basePower, suspended: true }, 1)).toBe(15_000); @@ -198,3 +224,206 @@ describe("commitCollectionControlUpdate", () => { }), ); }); + +describe("runPendingNativeTelemetryRequest", () => { + const makePending = () => + Ref.make>>(new Map()); + + const waitForPending = ( + pending: Ref.Ref>>, + ) => + Effect.gen(function* () { + while ((yield* Ref.get(pending)).size === 0) yield* Effect.yieldNow; + return [...(yield* Ref.get(pending)).values()][0]!; + }); + + it.effect("returns a direct response and removes the pending request", () => + Effect.gen(function* () { + const pending = yield* makePending(); + const fiber = yield* runPendingNativeTelemetryRequest({ + pending, + requestId: "response", + operation: "processTable", + timeout: Duration.seconds(5), + write: Effect.void, + }).pipe(Effect.forkChild); + yield* Deferred.succeed(yield* waitForPending(pending), 42); + + expect(yield* Fiber.join(fiber)).toBe(42); + expect((yield* Ref.get(pending)).size).toBe(0); + }), + ); + + it.effect("times out and removes the pending request", () => + Effect.gen(function* () { + const pending = yield* makePending(); + const fiber = yield* runPendingNativeTelemetryRequest({ + pending, + requestId: "timeout", + operation: "windowsListeners", + timeout: Duration.seconds(5), + write: Effect.void, + }).pipe(Effect.flip, Effect.forkChild); + yield* waitForPending(pending); + yield* TestClock.adjust(Duration.seconds(5)); + + expect((yield* Fiber.join(fiber))._tag).toBe("NativeTelemetryRequestTimedOut"); + expect((yield* Ref.get(pending)).size).toBe(0); + }), + ); + + it.effect("removes an interrupted pending request", () => + Effect.gen(function* () { + const pending = yield* makePending(); + const fiber = yield* runPendingNativeTelemetryRequest({ + pending, + requestId: "interrupted", + operation: "processTable", + timeout: Duration.seconds(5), + write: Effect.void, + }).pipe(Effect.forkChild); + yield* waitForPending(pending); + yield* Fiber.interrupt(fiber); + + expect((yield* Ref.get(pending)).size).toBe(0); + }), + ); + + it.effect("accepts a new request after a restart fails the old one", () => + Effect.gen(function* () { + const pending = yield* makePending(); + const first = yield* runPendingNativeTelemetryRequest({ + pending, + requestId: "before-restart", + operation: "processTable", + timeout: Duration.seconds(5), + write: Effect.void, + }).pipe(Effect.flip, Effect.forkChild); + yield* Deferred.fail( + yield* waitForPending(pending), + new NativeTelemetryExited({ exitCode: 1 }), + ); + expect((yield* Fiber.join(first))._tag).toBe("NativeTelemetryExited"); + + const second = yield* runPendingNativeTelemetryRequest({ + pending, + requestId: "after-restart", + operation: "processTable", + timeout: Duration.seconds(5), + write: Effect.void, + }).pipe(Effect.forkChild); + yield* Deferred.succeed(yield* waitForPending(pending), 7); + expect(yield* Fiber.join(second)).toBe(7); + expect((yield* Ref.get(pending)).size).toBe(0); + }), + ); +}); + +describe("NativeTelemetryClient", () => { + it.effect("fails an in-flight request when the sidecar exits and serves the replacement", () => + Effect.scoped( + Effect.gen(function* () { + const firstRequest = yield* Deferred.make(); + const firstReady = yield* Deferred.make(); + const secondSpawned = yield* Deferred.make(); + const secondReady = yield* Deferred.make(); + const firstExit = yield* Deferred.make(); + let spawnCount = 0; + + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const instance = spawnCount++; + const events = yield* Queue.unbounded(); + const hello = encodeMonitorEvent({ + version: RESOURCE_MONITOR_PROTOCOL_VERSION, + type: "hello", + sidecarVersion: "test", + sidecarPid: instance + 1, + platform: "test", + arch: "test", + capabilities: { + cumulativeCpuTime: true, + currentCpuPercent: true, + residentMemory: true, + virtualMemory: true, + ioBytes: true, + processStartTime: true, + processTree: true, + }, + }); + const stdin = Sink.forEach((chunk: Uint8Array) => + Effect.gen(function* () { + const command = decodeMonitorCommand(new TextDecoder().decode(chunk)); + if (command.type === "configure") { + yield* Deferred.succeed(instance === 0 ? firstReady : secondReady, undefined); + } + if (command.type !== "processTable") return; + if (instance === 0) { + yield* Deferred.succeed(firstRequest, undefined); + return; + } + yield* Queue.offer( + events, + new TextEncoder().encode( + `${encodeMonitorEvent({ + version: RESOURCE_MONITOR_PROTOCOL_VERSION, + type: "processTable", + requestId: command.requestId, + processes: [{ pid: 42, ppid: 1, name: "replacement-child" }], + })}\n`, + ), + ); + }), + ); + if (instance === 1) yield* Deferred.succeed(secondSpawned, undefined); + return ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(instance + 1), + exitCode: instance === 0 ? Deferred.await(firstExit) : Effect.never, + isRunning: Effect.succeed(true), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin, + stdout: Stream.concat( + Stream.encodeText(Stream.make(`${hello}\n`)), + Stream.fromQueue(events), + ), + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }); + }), + ); + const dependencies = Layer.mergeAll( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-native-telemetry-client-test-" }), + Layer.succeed( + ResourceMonitorBinary.ResourceMonitorBinary, + ResourceMonitorBinary.ResourceMonitorBinary.of({ + resolve: Effect.succeed("test-sidecar"), + }), + ), + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + ).pipe(Layer.provideMerge(NodeServices.layer)); + const context = yield* Layer.build( + nativeTelemetryClientLayer.pipe(Layer.provide(dependencies)), + ); + const client = yield* Effect.service(NativeTelemetryClient).pipe(Effect.provide(context)); + + yield* Deferred.await(firstReady); + yield* client.capabilities; + const first = yield* client.processTable.pipe(Effect.flip, Effect.forkChild); + yield* Deferred.await(firstRequest); + yield* Deferred.succeed(firstExit, ChildProcessSpawner.ExitCode(1)); + expect((yield* Fiber.join(first))._tag).toBe("NativeTelemetryExited"); + + yield* TestClock.adjust(Duration.millis(500)); + yield* Deferred.await(secondSpawned); + yield* Deferred.await(secondReady); + + expect(yield* client.processTable).toEqual([ + { pid: 42, ppid: 1, name: "replacement-child" }, + ]); + }), + ), + ); +}); diff --git a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts index 40254bfe80aa..f7b74396a0de 100644 --- a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts +++ b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts @@ -7,6 +7,7 @@ import type { ResourceMonitorHelloEvent, ResourceMonitorProcessTableEntry, ResourceMonitorSnapshotEvent, + ResourceMonitorWindowsListener, ResourceTelemetrySourceStatus, } from "@t3tools/contracts"; import { @@ -78,7 +79,7 @@ export class NativeTelemetryHandshakeTimedOut extends Schema.TaggedError()( "NativeTelemetryRequestTimedOut", { - operation: Schema.Literals(["processTable", "readHistory", "sampleNow"]), + operation: Schema.Literals(["processTable", "readHistory", "sampleNow", "windowsListeners"]), timeoutMs: Schema.Number, }, ) { @@ -198,6 +199,10 @@ export class NativeTelemetryClient extends Context.Service< ReadonlyArray, NativeTelemetryClientError >; + readonly windowsListeners: Effect.Effect< + ReadonlyArray, + NativeTelemetryClientError + >; readonly retry: Effect.Effect; readonly health: Effect.Effect; readonly subscribeHealth: Effect.Effect< @@ -360,6 +365,50 @@ export function canCommandNativeTelemetrySidecar( return hasHandle && (status === "healthy" || status === "degraded"); } +export function runPendingNativeTelemetryRequest(input: { + readonly pending: Ref.Ref>>; + readonly requestId: string; + readonly operation: "processTable" | "sampleNow" | "windowsListeners"; + readonly timeout: Duration.Duration; + readonly write: Effect.Effect; +}): Effect.Effect { + return Effect.gen(function* () { + const deferred = yield* Deferred.make(); + yield* Ref.update(input.pending, (pending) => { + const next = new Map(pending); + next.set(input.requestId, deferred); + return next; + }); + return yield* input.write.pipe( + Effect.andThen( + Deferred.await(deferred).pipe( + Effect.timeoutOption(input.timeout), + Effect.flatMap( + Option.match({ + onNone: () => + Effect.fail( + new NativeTelemetryRequestTimedOut({ + operation: input.operation, + timeoutMs: Duration.toMillis(input.timeout), + }), + ), + onSome: Effect.succeed, + }), + ), + ), + ), + ); + }).pipe( + Effect.ensuring( + Ref.update(input.pending, (pending) => { + const next = new Map(pending); + next.delete(input.requestId); + return next; + }), + ), + ); +} + /** @public Service construction is part of the canonical Effect module API. */ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(function* () { const binary = yield* ResourceMonitorBinary.ResourceMonitorBinary; @@ -395,6 +444,12 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu Deferred.Deferred, NativeTelemetryClientError> >(), ); + const pendingWindowsListeners = 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); @@ -413,6 +468,7 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu Effect.gen(function* () { const samples = yield* Ref.getAndSet(pendingSamples, new Map()); const processTables = yield* Ref.getAndSet(pendingProcessTables, new Map()); + const windowsListeners = yield* Ref.getAndSet(pendingWindowsListeners, new Map()); const histories = yield* Ref.getAndSet(pendingHistories, new Map()); yield* Effect.forEach(samples.values(), (deferred) => Deferred.fail(deferred, error), { discard: true, @@ -420,6 +476,11 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu yield* Effect.forEach(processTables.values(), (deferred) => Deferred.fail(deferred, error), { discard: true, }); + yield* Effect.forEach( + windowsListeners.values(), + (deferred) => Deferred.fail(deferred, error), + { discard: true }, + ); yield* Effect.forEach( histories.values(), (request) => Deferred.fail(request.deferred, error), @@ -513,6 +574,30 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu ), Effect.asVoid, ); + case "windowsListeners": + return Ref.modify(pendingWindowsListeners, (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) => + event.error === null + ? Deferred.succeed(deferred, event.listeners) + : Deferred.fail( + deferred, + new NativeTelemetryCommandFailed({ + operation: "windowsListeners", + cause: event.error, + }), + ), + }), + ), + Effect.asVoid, + ); case "historyChunk": return Effect.gen(function* () { const latestSnapshot = event.snapshots.at(-1); @@ -930,42 +1015,17 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu }), ), ); - const deferred = yield* Deferred.make(); - yield* Ref.update(pendingSamples, (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: "sampleNow", + return yield* runPendingNativeTelemetryRequest({ + pending: pendingSamples, requestId, - }).pipe( - Effect.andThen( - Deferred.await(deferred).pipe( - Effect.timeoutOption(SAMPLE_REQUEST_TIMEOUT), - Effect.flatMap( - Option.match({ - onNone: () => - Effect.fail( - new NativeTelemetryRequestTimedOut({ - operation: "sampleNow", - timeoutMs: Duration.toMillis(SAMPLE_REQUEST_TIMEOUT), - }), - ), - onSome: Effect.succeed, - }), - ), - ), - ), - Effect.ensuring( - Ref.update(pendingSamples, (pending) => { - const next = new Map(pending); - next.delete(requestId); - return next; - }), - ), - ); + operation: "sampleNow", + timeout: SAMPLE_REQUEST_TIMEOUT, + write: writeCommand(Option.getOrThrow(current.handle), { + version: RESOURCE_MONITOR_PROTOCOL_VERSION, + type: "sampleNow", + requestId, + }), + }); }); const processTable: NativeTelemetryClient["Service"]["processTable"] = Effect.gen(function* () { @@ -981,46 +1041,46 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu (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", + return yield* runPendingNativeTelemetryRequest({ + pending: pendingProcessTables, 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, - }), - ), + operation: "processTable", + timeout: PROCESS_TABLE_REQUEST_TIMEOUT, + write: writeCommand(Option.getOrThrow(current.handle), { + version: RESOURCE_MONITOR_PROTOCOL_VERSION, + type: "processTable", + requestId, + }), + }); + }); + + const windowsListeners: NativeTelemetryClient["Service"]["windowsListeners"] = 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 }), ), - ), - Effect.ensuring( - Ref.update(pendingProcessTables, (pending) => { - const next = new Map(pending); - next.delete(requestId); - return next; + ); + return yield* runPendingNativeTelemetryRequest({ + pending: pendingWindowsListeners, + requestId, + operation: "windowsListeners", + timeout: PROCESS_TABLE_REQUEST_TIMEOUT, + write: writeCommand(Option.getOrThrow(current.handle), { + version: RESOURCE_MONITOR_PROTOCOL_VERSION, + type: "windowsListeners", + requestId, }), - ), - ); - }); + }); + }, + ); const health = currentHealth; @@ -1044,6 +1104,7 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu setHostPowerState, sampleNow, processTable, + windowsListeners, retry: Ref.get(state).pipe( Effect.flatMap((current) => !canRequestNativeTelemetryRetry(current.status, Option.isSome(current.handle)) @@ -1102,6 +1163,11 @@ export const layerTest = ( reason: "No resource monitor process table was configured for this test.", }), ), + windowsListeners: Effect.fail( + new NativeTelemetryUnavailable({ + reason: "No Windows listener 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 9c371078332d..62e22b31a785 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: 3, + version: 4, 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: 3, + version: 4, 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 fff4588c8468..1f7181adac2f 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: 3, + version: 4, type: "snapshot", sequence, sampledAtUnixMs, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index fd8ee4a4f699..437cd2a5a304 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -369,7 +369,10 @@ const CheckpointingLayerLive = Layer.empty.pipe( Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistryLayerLive))), ); -const PortScannerLayerLive = PortScanner.layer.pipe(Layer.provide(ProcessRunner.layer)); +const PortScannerLayerLive = PortScanner.layer.pipe( + Layer.provide(ProcessRunner.layer), + Layer.provide(NativeTelemetryLayerLive), +); const TerminalLayerLive = TerminalManager.layer.pipe( Layer.provide(PtyAdapterLive), diff --git a/apps/server/src/terminal/Manager.test.ts b/apps/server/src/terminal/Manager.test.ts index f631992e7ae3..44803c94b093 100644 --- a/apps/server/src/terminal/Manager.test.ts +++ b/apps/server/src/terminal/Manager.test.ts @@ -1203,6 +1203,8 @@ it.layer( 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); + assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 1, 3_000), 12_000); + assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 1, 30_000), 60_000); }); it.effect("uses process snapshots from the resource monitor", () => diff --git a/apps/server/src/terminal/Manager.ts b/apps/server/src/terminal/Manager.ts index 174ed4206afc..5610a2ad6b50 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -647,8 +647,12 @@ interface TerminalProcessTableSnapshot { export function subprocessSnapshotPollDelayMs( pollIntervalMs: number, failureCount: number, + fallbackElapsedMs = 0, ): number { - return Math.min(pollIntervalMs * 2 ** failureCount, MAX_SUBPROCESS_POLL_INTERVAL_MS); + return Math.min( + Math.max(pollIntervalMs * 2 ** failureCount, Math.max(0, fallbackElapsedMs) * 4), + MAX_SUBPROCESS_POLL_INTERVAL_MS, + ); } function parsePosixProcessTable(stdout: string): TerminalProcessTableSnapshot { @@ -780,7 +784,7 @@ const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnaps .run({ command: "powershell.exe", args: ["-NoProfile", "-NonInteractive", "-Command", command], - timeout: "1500 millis", + timeout: "15 seconds", maxOutputBytes: 262_144, outputMode: "truncate", timeoutBehavior: "timedOutResult", @@ -1471,6 +1475,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func * a failure so polling backs off instead of hot-looping the fallback. */ readonly snapshotSucceeded: boolean; + readonly fallbackElapsedMs: number; }, TerminalSubprocessCheckError > = options.processTable @@ -1478,38 +1483,54 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func Effect.map((entries) => ({ snapshot: processTableSnapshotFromProcesses(entries), snapshotSucceeded: true, + fallbackElapsedMs: 0, })), Effect.catch(() => - fallbackProcessTableSnapshot.pipe( - Effect.map((snapshot) => ({ snapshot, snapshotSucceeded: false })), - ), + Effect.suspend(() => { + const startedAtMillis = performance.now(); + return fallbackProcessTableSnapshot.pipe( + Effect.map((snapshot) => ({ + snapshot, + snapshotSucceeded: false, + fallbackElapsedMs: performance.now() - startedAtMillis, + })), + ); + }), ), ) : fallbackProcessTableSnapshot.pipe( - Effect.map((snapshot) => ({ snapshot, snapshotSucceeded: true })), + Effect.map((snapshot) => ({ snapshot, snapshotSucceeded: true, fallbackElapsedMs: 0 })), ); const customSubprocessInspector = options.subprocessInspector; const acquireSubprocessInspector: Effect.Effect< { readonly inspector: TerminalSubprocessInspector; readonly snapshotSucceeded: boolean; + readonly fallbackElapsedMs: number; }, TerminalSubprocessCheckError > = customSubprocessInspector !== undefined - ? Effect.succeed({ inspector: customSubprocessInspector, snapshotSucceeded: true }) + ? Effect.succeed({ + inspector: customSubprocessInspector, + snapshotSucceeded: true, + fallbackElapsedMs: 0, + }) : Effect.map( fetchProcessTableSnapshot, ({ snapshot, snapshotSucceeded, + fallbackElapsedMs, }): { readonly inspector: TerminalSubprocessInspector; readonly snapshotSucceeded: boolean; + readonly fallbackElapsedMs: number; } => ({ inspector: (terminalPid) => Effect.succeed(deriveSubprocessInspectResult(snapshot, terminalPid, platform)), snapshotSucceeded, + fallbackElapsedMs, }), ); const subprocessPollIntervalMs = @@ -2362,7 +2383,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func ); if (runningSessions.length === 0) { - return true; + return { snapshotSucceeded: true, fallbackElapsedMs: 0 }; } const inspectorOption = yield* acquireSubprocessInspector.pipe( @@ -2375,6 +2396,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func Option.none<{ readonly inspector: TerminalSubprocessInspector; readonly snapshotSucceeded: boolean; + readonly fallbackElapsedMs: number; }>(), ), ), @@ -2382,10 +2404,14 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func ); if (Option.isNone(inspectorOption)) { - return false; + return { snapshotSucceeded: false, fallbackElapsedMs: 0 }; } - const { inspector: subprocessInspector, snapshotSucceeded } = inspectorOption.value; + const { + inspector: subprocessInspector, + snapshotSucceeded, + fallbackElapsedMs, + } = inspectorOption.value; const checkSubprocessActivity = Effect.fn("terminal.checkSubprocessActivity")(function* ( session: TerminalSessionState & { pid: number }, @@ -2454,7 +2480,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func concurrency: "unbounded", discard: true, }); - return snapshotSucceeded; + return { snapshotSucceeded, fallbackElapsedMs }; }); const hasRunningSessions = readManagerState.pipe( @@ -2469,13 +2495,14 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func Effect.flatMap((active) => active ? pollSubprocessActivity().pipe( - Effect.flatMap((snapshotSucceeded) => { + Effect.flatMap(({ snapshotSucceeded, fallbackElapsedMs }) => { subprocessSnapshotFailureCount = snapshotSucceeded ? 0 : Math.min(subprocessSnapshotFailureCount + 1, 30); const delayMs = subprocessSnapshotPollDelayMs( subprocessPollIntervalMs, subprocessSnapshotFailureCount, + fallbackElapsedMs, ); return Effect.sleep(delayMs); }), diff --git a/native/resource-monitor/src/main.rs b/native/resource-monitor/src/main.rs index 6137544ea0e2..bebc71622134 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 = 3; +const PROTOCOL_VERSION: u32 = 4; const MIN_SAMPLE_INTERVAL_MS: u64 = 250; const MAX_SAMPLE_INTERVAL_MS: u64 = 60_000; const PROCESS_START_TIME_PRECISION_MS: u64 = 1_000; @@ -70,6 +70,10 @@ enum Command { version: u32, request_id: String, }, + WindowsListeners { + version: u32, + request_id: String, + }, ReadHistory { version: u32, request_id: String, @@ -89,6 +93,7 @@ impl Command { | Self::SetStreaming { version, .. } | Self::SampleNow { version, .. } | Self::ProcessTable { version, .. } + | Self::WindowsListeners { version, .. } | Self::ReadHistory { version, .. } | Self::Shutdown { version } => *version, } @@ -169,6 +174,159 @@ struct ProcessTableEvent<'a> { processes: Vec, } +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +struct WindowsListener { + port: u16, + pid: u32, + process_name: Option, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct WindowsListenersEvent<'a> { + version: u32, + #[serde(rename = "type")] + event_type: &'static str, + request_id: &'a str, + listeners: Vec, + error: Option, +} + +#[cfg(target_os = "windows")] +mod windows_listeners { + use std::ffi::c_void; + use std::io; + use std::mem::size_of; + use std::ptr; + + const AF_INET: u32 = 2; + const AF_INET6: u32 = 23; + const ERROR_INSUFFICIENT_BUFFER: u32 = 122; + const MAX_TABLE_READ_ATTEMPTS: usize = 3; + const TCP_TABLE_OWNER_PID_LISTENER: u32 = 3; + + #[repr(C)] + #[derive(Clone, Copy)] + struct TcpRow { + state: u32, + local_address: u32, + local_port: u32, + remote_address: u32, + remote_port: u32, + owning_pid: u32, + } + + #[repr(C)] + #[derive(Clone, Copy)] + struct Tcp6Row { + local_address: [u8; 16], + local_scope_id: u32, + local_port: u32, + remote_address: [u8; 16], + remote_scope_id: u32, + remote_port: u32, + state: u32, + owning_pid: u32, + } + + #[link(name = "iphlpapi")] + unsafe extern "system" { + fn GetExtendedTcpTable( + table: *mut c_void, + size: *mut u32, + order: i32, + family: u32, + table_class: u32, + reserved: u32, + ) -> u32; + } + + unsafe fn rows(family: u32) -> io::Result> { + let mut size = 0u32; + let first = unsafe { + GetExtendedTcpTable( + ptr::null_mut(), + &mut size, + 0, + family, + TCP_TABLE_OWNER_PID_LISTENER, + 0, + ) + }; + if first != ERROR_INSUFFICIENT_BUFFER { + return Err(io::Error::from_raw_os_error(first as i32)); + } + for _ in 0..MAX_TABLE_READ_ATTEMPTS { + let word_count = (size as usize).div_ceil(size_of::()); + let mut buffer = vec![0u32; word_count]; + let result = unsafe { + GetExtendedTcpTable( + buffer.as_mut_ptr().cast(), + &mut size, + 0, + family, + TCP_TABLE_OWNER_PID_LISTENER, + 0, + ) + }; + if result == ERROR_INSUFFICIENT_BUFFER { + continue; + } + if result != 0 { + return Err(io::Error::from_raw_os_error(result as i32)); + } + let count = unsafe { ptr::read_unaligned(buffer.as_ptr()) } as usize; + let available_bytes = + (buffer.len() * size_of::()).saturating_sub(size_of::()); + if count > available_bytes / size_of::() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Windows TCP table exceeded its returned buffer", + )); + } + let first_row = unsafe { buffer.as_ptr().cast::().add(size_of::()) }; + return Ok((0..count) + .map(|index| unsafe { + ptr::read_unaligned(first_row.add(index * size_of::()).cast::()) + }) + .collect()); + } + Err(io::Error::new( + io::ErrorKind::WouldBlock, + "Windows TCP table kept growing while it was read", + )) + } + + fn port(value: u32) -> u16 { + u16::from_be(value as u16) + } + + pub fn read() -> io::Result> { + let ipv4 = unsafe { rows::(AF_INET)? }; + let ipv6 = unsafe { rows::(AF_INET6)? }; + let mut listeners = ipv4 + .into_iter() + .filter(|row| { + row.local_address == 0 || row.local_address.to_ne_bytes().first() == Some(&127) + }) + .map(|row| (port(row.local_port), row.owning_pid)) + .chain( + ipv6.into_iter() + .filter(|row| { + row.local_address.iter().all(|byte| *byte == 0) + || row.local_address == [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1] + }) + .map(|row| (port(row.local_port), row.owning_pid)), + ) + .filter(|(port, pid)| *port > 0 && *pid > 0) + .collect::>(); + listeners.sort_unstable(); + listeners.dedup(); + Ok(listeners) + } +} + impl ProcessSample { fn estimated_history_bytes(&self) -> usize { std::mem::size_of::() @@ -409,6 +567,32 @@ impl Collector { processes } + #[cfg(target_os = "windows")] + fn windows_listeners(&mut self) -> Result, String> { + let process_names = self + .process_table() + .into_iter() + .map(|process| (process.pid, process.name)) + .collect::>(); + windows_listeners::read() + .map(|listeners| { + listeners + .into_iter() + .map(|(port, pid)| WindowsListener { + port, + pid, + process_name: process_names.get(&pid).cloned(), + }) + .collect() + }) + .map_err(|error| error.to_string()) + } + + #[cfg(not(target_os = "windows"))] + fn windows_listeners(&mut self) -> Result, String> { + Err("Windows listener discovery is unavailable on this platform".to_owned()) + } + 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()) @@ -938,6 +1122,17 @@ fn main() -> io::Result<()> { }; write_event(&mut writer, &event)?; } + Command::WindowsListeners { request_id, .. } => { + let result = collector.windows_listeners(); + let event = WindowsListenersEvent { + version: PROTOCOL_VERSION, + event_type: "windowsListeners", + request_id: &request_id, + listeners: result.as_ref().cloned().unwrap_or_default(), + error: result.err(), + }; + write_event(&mut writer, &event)?; + } Command::ReadHistory { request_id, window_ms, @@ -968,6 +1163,13 @@ fn main() -> io::Result<()> { mod tests { use super::*; + #[cfg(target_os = "windows")] + #[test] + fn reads_windows_listener_table() { + let listeners = windows_listeners::read().expect("Windows listener table"); + assert!(listeners.iter().all(|(port, pid)| *port > 0 && *pid > 0)); + } + #[test] fn selects_roots_and_all_descendants() { let rows = vec![ @@ -1048,7 +1250,7 @@ mod tests { #[test] fn decodes_protocol_commands() { let configure = serde_json::from_str::( - r#"{"version":3,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#, + r#"{"version":4,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#, ) .expect("configure command"); @@ -1068,7 +1270,7 @@ mod tests { } let read_history = serde_json::from_str::( - r#"{"version":3,"type":"readHistory","requestId":"history-1","windowMs":60000}"#, + r#"{"version":4,"type":"readHistory","requestId":"history-1","windowMs":60000}"#, ) .expect("read history command"); assert!(matches!( @@ -1081,13 +1283,22 @@ mod tests { )); let process_table = serde_json::from_str::( - r#"{"version":3,"type":"processTable","requestId":"processes-1"}"#, + r#"{"version":4,"type":"processTable","requestId":"processes-1"}"#, ) .expect("process table command"); assert!(matches!( process_table, Command::ProcessTable { request_id, .. } if request_id == "processes-1" )); + + let windows_listeners = serde_json::from_str::( + r#"{"version":4,"type":"windowsListeners","requestId":"listeners-1"}"#, + ) + .expect("Windows listeners command"); + assert!(matches!( + windows_listeners, + Command::WindowsListeners { request_id, .. } if request_id == "listeners-1" + )); } #[test] diff --git a/packages/contracts/src/resourceTelemetry.ts b/packages/contracts/src/resourceTelemetry.ts index ee9d2b3ac258..84d7e71e4634 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 = 3 as const; +export const RESOURCE_MONITOR_PROTOCOL_VERSION = 4 as const; /** Whole-host capacity, independent of T3's process diagnostics. */ export const HostResourcesSnapshot = Schema.Struct({ @@ -119,6 +119,14 @@ export const ResourceMonitorProcessTableCommand = Schema.Struct({ }); export type ResourceMonitorProcessTableCommand = typeof ResourceMonitorProcessTableCommand.Type; +export const ResourceMonitorWindowsListenersCommand = Schema.Struct({ + version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), + type: Schema.Literal("windowsListeners"), + requestId: TrimmedNonEmptyString, +}); +export type ResourceMonitorWindowsListenersCommand = + typeof ResourceMonitorWindowsListenersCommand.Type; + export const ResourceMonitorSetSampleIntervalCommand = Schema.Struct({ version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), type: Schema.Literal("setSampleInterval"), @@ -155,6 +163,7 @@ export const ResourceMonitorCommand = Schema.Union([ ResourceMonitorSetStreamingCommand, ResourceMonitorSampleNowCommand, ResourceMonitorProcessTableCommand, + ResourceMonitorWindowsListenersCommand, ResourceMonitorReadHistoryCommand, ResourceMonitorShutdownCommand, ]); @@ -201,6 +210,22 @@ export const ResourceMonitorProcessTableEvent = Schema.Struct({ }); export type ResourceMonitorProcessTableEvent = typeof ResourceMonitorProcessTableEvent.Type; +export const ResourceMonitorWindowsListener = Schema.Struct({ + port: PositiveInt, + pid: PositiveInt, + processName: Schema.NullOr(Schema.String), +}); +export type ResourceMonitorWindowsListener = typeof ResourceMonitorWindowsListener.Type; + +export const ResourceMonitorWindowsListenersEvent = Schema.Struct({ + version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), + type: Schema.Literal("windowsListeners"), + requestId: TrimmedNonEmptyString, + listeners: Schema.Array(ResourceMonitorWindowsListener), + error: Schema.NullOr(Schema.String), +}); +export type ResourceMonitorWindowsListenersEvent = typeof ResourceMonitorWindowsListenersEvent.Type; + export const ResourceMonitorHistoryChunkEvent = Schema.Struct({ version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION), type: Schema.Literal("historyChunk"), @@ -223,6 +248,7 @@ export const ResourceMonitorEvent = Schema.Union([ ResourceMonitorHelloEvent, ResourceMonitorSnapshotEvent, ResourceMonitorProcessTableEvent, + ResourceMonitorWindowsListenersEvent, ResourceMonitorHistoryChunkEvent, ResourceMonitorErrorEvent, ]);