Skip to content
134 changes: 134 additions & 0 deletions apps/server/src/preview/PortScanner.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import {
} from "@t3tools/contracts";
import * as HostProcess from "@t3tools/shared/HostProcess";
import * as Cause from "effect/Cause";
import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
import * as Clock from "effect/Clock";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
Expand Down Expand Up @@ -789,3 +790,136 @@ effectIt.effect("does not swallow process probe interruption", () =>
}
}),
);

effectIt.effect(
"parses Windows netstat listeners, names them via tasklist with retry, and skips spawns when idle",
() => {
const spawnedArgs: Array<ReadonlyArray<string>> = [];
let tasklistCalls = 0;
const fetchFn = (() =>
Promise.resolve(
new Response("hello", { headers: { "content-type": "text/html" } }),
)) as typeof globalThis.fetch;
const layer = PortScanner.layer.pipe(
Layer.provide(
Layer.mergeAll(
layerNoProc,
Layer.succeed(ProcessRunner.ProcessRunner, {
run: (input) => {
spawnedArgs.push(input.args);
const isIpv6 = input.args.includes("tcpv6");
const isTasklist = input.command === "tasklist.exe";
// The first lookup fails, so it must not be cached as "no name".
const failTasklist = isTasklist && tasklistCalls++ === 0;
return Effect.succeed({
stdout: isTasklist
? '"node.exe","1234","Console","1","93,164 K"\r\n'
: isIpv6
? " TCP [::1]:43123 [::]:0 ABHÖREN 1234\r\n TCP [::1]:43123 [::1]:52000 HERGESTELLT 1234\r\n"
: " TCP 127.0.0.1:5173 0.0.0.0:0 LISTENING 1234\r\n TCP 127.0.0.1:5173 127.0.0.1:60123 ESTABLISHED 9999\r\n",
stderr: "",
code: failTasklist ? ChildProcessSpawner.ExitCode(1) : null,
timedOut: false,
stdoutTruncated: false,
stderrTruncated: false,
stdoutInvalidUtf8: false,
stderrInvalidUtf8: false,
});
},
}),
Layer.succeed(HostProcess.Platform, "win32"),
FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetchFn))),
),
),
);

return Effect.gen(function* () {
const scanner = yield* PortScanner.PortDiscovery;
expect(yield* scanner.scan()).toEqual([]);
expect(spawnedArgs).toHaveLength(0);

yield* scanner.registerTerminalProcesses({
threadId: "scanner-thread",
terminalId: "scanner-terminal",
processIds: [1234],
});
const summarize = (
servers: ReadonlyArray<{
readonly port: number;
readonly pid: number | null;
readonly processName: string | null;
}>,
) =>
servers.map((entry) => ({
port: entry.port,
pid: entry.pid,
processName: entry.processName,
}));
expect(summarize(yield* scanner.scan())).toEqual([
{ port: 5173, pid: 1234, processName: null },
{ port: 43123, pid: 1234, processName: null },
]);
expect(spawnedArgs).toEqual([
["-ano", "-p", "tcp"],
["-ano", "-p", "tcpv6"],
["/FO", "CSV", "/NH", "/FI", "PID eq 1234"],
]);

// The failed lookup retries; the result is then cached, so later polls spawn only netstat.
expect(summarize(yield* scanner.scan())).toEqual([
{ port: 5173, pid: 1234, processName: "node" },
{ port: 43123, pid: 1234, processName: "node" },
]);
yield* scanner.scan();
expect(tasklistCalls).toBe(2);

// Cached names expire, since Windows may hand the PID to another process.
yield* TestClock.adjust(Duration.seconds(61));
yield* scanner.scan();
expect(tasklistCalls).toBe(3);
}).pipe(Effect.provide(layer));
},
);

effectIt.effect("ignores a partial Windows netstat snapshot when one query exits nonzero", () => {
const fetchFn = (() =>
Promise.resolve(
new Response("hello", { headers: { "content-type": "text/html" } }),
)) as typeof globalThis.fetch;
const layer = PortScanner.layer.pipe(
Layer.provide(
Layer.mergeAll(
layerNoProc,
Layer.succeed(ProcessRunner.ProcessRunner, {
run: (input) => {
const isIpv6 = input.args.includes("tcpv6");
return Effect.succeed({
stdout: isIpv6
? ""
: " TCP 127.0.0.1:5173 0.0.0.0:0 LISTENING 1234\r\n",
stderr: "",
code: isIpv6 ? ChildProcessSpawner.ExitCode(1) : null,
timedOut: false,
stdoutTruncated: false,
stderrTruncated: false,
stdoutInvalidUtf8: false,
stderrInvalidUtf8: false,
});
},
}),
Layer.succeed(HostProcess.Platform, "win32"),
FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetchFn))),
),
),
);

return Effect.gen(function* () {
const scanner = yield* PortScanner.PortDiscovery;
yield* scanner.registerTerminalProcesses({
threadId: "scanner-thread",
terminalId: "scanner-terminal",
processIds: [1234],
});
expect(yield* scanner.scan()).toEqual([]);
}).pipe(Effect.provide(layer));
});
185 changes: 157 additions & 28 deletions apps/server/src/preview/PortScanner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -77,8 +77,13 @@ const PROC_FD_WALK_LIMIT = 50_000;
/** Scans that retry the fd walk for a listening socket whose owner was not found. */
const SOCKET_OWNER_ATTEMPTS = 3;
const WINDOWS_LISTENER_TIMEOUT_MS = 5_000;
/** `netstat -ano` lists every connection, not just listeners; ~80 bytes per row, sized for ~200k rows. */
const WINDOWS_NETSTAT_MAX_OUTPUT_BYTES = 16 * 1024 * 1024;
const WEB_PROBE_TIMEOUT = Duration.seconds(1);
const WEB_PROBE_CACHE_TTL_MS = Duration.toMillis(Duration.seconds(15));
/** Bounds how long a cached Windows image name can outlive its process if Windows reuses the PID. */
const WINDOWS_PROCESS_NAME_TTL_MS = Duration.toMillis(Duration.seconds(60));
const WINDOWS_PROCESS_NAME_CONCURRENCY = 4;
const WEB_PROBE_CONCURRENCY = 16;
const NAVIGATION_REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);

Expand Down Expand Up @@ -249,19 +254,78 @@ const parsePortFromLsofName = (name: string): number | null => {
return port;
};

const parseWindowsListenerLine = (
line: string,
): {
readonly host: string;
readonly port: number;
readonly pid: number | null;
readonly processName: string | null;
} | null => {
const trimmed = line.trim();
if (trimmed.length === 0) return null;

if (trimmed.includes("|")) {
const [hostRaw, portRaw, pidRaw, processNameRaw] = trimmed.split("|", 4);
const host = hostRaw?.trim() ?? "";
if (!LSOF_LOCAL_HOST_TOKENS.has(host) && host !== "::") return null;
const port = Number(portRaw);
const pid = Number(pidRaw);
if (!Number.isInteger(port) || port <= 0 || port >= 65536) return null;
return {
host,
port,
pid: Number.isInteger(pid) && pid > 0 ? pid : null,
processName: processNameRaw?.trim() || null,
};
}

// `netstat.exe -ano -p tcp` / `tcpv6`:
// TCP 127.0.0.1:3773 0.0.0.0:0 LISTENING 34280
// TCP [::1]:35783 [::]:0 LISTENING 22876
// Foreign port `:0` identifies a listening TCP socket regardless of Windows locale.
const fields = trimmed.split(/\s+/);
if (fields.length < 4 || fields[0]?.toUpperCase() !== "TCP") return null;
const localEndpoint = fields[1] ?? "";
const foreignEndpoint = fields[2] ?? "";
if (!foreignEndpoint.endsWith(":0")) return null;
const lastColon = localEndpoint.lastIndexOf(":");
if (lastColon <= 0) return null;
const host = localEndpoint.slice(0, lastColon);
if (!LSOF_LOCAL_HOST_TOKENS.has(host) && host !== "::") return null;
const port = Number(localEndpoint.slice(lastColon + 1));
if (!Number.isInteger(port) || port <= 0 || port >= 65536) return null;
const pid = Number(fields.at(-1));
return {
host,
port,
pid: Number.isInteger(pid) && pid > 0 ? pid : null,
processName: null,
Comment thread
mweinbach marked this conversation as resolved.
};
};

/** `ProcessRunner` reports a nonzero exit as a result, not a failure. */
const exitedWithError = (result: ProcessRunner.ProcessRunOutput): boolean =>
result.code !== null && result.code !== 0;

/** Reads the image name from `tasklist.exe /FO CSV /NH /FI "PID eq <pid>"` output. */
const parseTasklistProcessName = (raw: string, pid: number): string | null => {
for (const line of raw.split(/\r?\n/g)) {
const match = /^"([^"]*)","(\d+)"/.exec(line.trim());
if (match && Number(match[2]) === pid) return match[1]!.replace(/\.exe$/i, "") || null;
}
return null;
};

const parseWindowsListenerOutput = (
raw: string,
terminalByProcessId: ReadonlyMap<number, TerminalProcessOwner> = new Map(),
): ReadonlyArray<DiscoveredLocalServer> => {
const seen = new Map<number, DiscoveredLocalServer>();
for (const line of raw.split(/\r?\n/g)) {
const [hostRaw, portRaw, pidRaw, processNameRaw] = line.trim().split("|", 4);
const host = hostRaw?.trim() ?? "";
if (!LSOF_LOCAL_HOST_TOKENS.has(host) && host !== "::") continue;
const port = Number(portRaw);
const pid = Number(pidRaw);
if (!Number.isInteger(port) || port <= 0 || port >= 65536) continue;
const normalizedPid = Number.isInteger(pid) && pid > 0 ? pid : null;
const parsed = parseWindowsListenerLine(line);
if (!parsed) continue;
const { port, pid: normalizedPid, processName } = parsed;
const terminal =
normalizedPid === null ? null : (terminalByProcessId.get(normalizedPid) ?? null);
const existing = seen.get(port);
Expand All @@ -273,7 +337,7 @@ const parseWindowsListenerOutput = (
host: "localhost",
port,
url: `http://localhost:${port}`,
processName: processNameRaw?.trim() || null,
processName,
pid: normalizedPid,
terminal,
});
Expand Down Expand Up @@ -358,6 +422,10 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
const webProbeCacheRef = yield* Ref.make<ReadonlyMap<string, WebProbeCacheEntry>>(new Map());
const scanSemaphore = yield* Semaphore.make(1);
const lsofMissingRef = yield* Ref.make(false);
/** Windows image names by PID; netstat reports only PIDs. */
const windowsProcessNamesRef = yield* Ref.make<
ReadonlyMap<number, { readonly name: string | null; readonly expiresAtMillis: number }>
>(new Map());
/** Confirmed owners of listening socket inodes. */
const socketOwnersRef = yield* Ref.make<
ReadonlyMap<string, { readonly pid: number; readonly processName: string | null }>
Expand Down Expand Up @@ -604,6 +672,57 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
return { discovered, configured } satisfies WebProbeSnapshot;
});

// `tasklist` costs ~200-300ms per PID, so look up only visible servers, a few at a time, and
// reuse each name until its TTL expires.
const withWindowsProcessNames = Effect.fn("PortDiscovery.withWindowsProcessNames")(function* (
snapshot: WebProbeSnapshot,
) {
const servers = [...snapshot.discovered, ...snapshot.configured.values()];
const pids = new Set(servers.flatMap((server) => (server.pid === null ? [] : [server.pid])));
const nowMillis = yield* Clock.currentTimeMillis;
const cached = [...(yield* Ref.get(windowsProcessNamesRef))].filter(
([pid, entry]) => pids.has(pid) && entry.expiresAtMillis > nowMillis,
);
const cachedPids = new Set(cached.map(([pid]) => pid));
const expiresAtMillis = nowMillis + WINDOWS_PROCESS_NAME_TTL_MS;
const fetched = yield* Effect.forEach(
[...pids].filter((pid) => !cachedPids.has(pid)),
(pid) =>
processRunner
.run({
command: "tasklist.exe",
args: ["/FO", "CSV", "/NH", "/FI", `PID eq ${pid}`],
timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS),
maxOutputBytes: 64 * 1024,
})
.pipe(
// Only a completed lookup is cached; failures retry on the next scan.
Effect.map((result) =>
exitedWithError(result)
? []
: [
[
pid,
{ name: parseTasklistProcessName(result.stdout, pid), expiresAtMillis },
] as const,
],
),
Effect.orElseSucceed(() => []),
),
{ concurrency: WINDOWS_PROCESS_NAME_CONCURRENCY },
);
const names = new Map([...cached, ...fetched.flat()]);
yield* Ref.set(windowsProcessNamesRef, names);
const withName = (server: DiscoveredLocalServer): DiscoveredLocalServer =>
server.pid === null || server.processName !== null
? server
: { ...server, processName: names.get(server.pid)?.name ?? null };
return {
discovered: snapshot.discovered.map(withName),
configured: new Map([...snapshot.configured].map(([key, server]) => [key, withName(server)])),
} satisfies WebProbeSnapshot;
});

const recoverProcessProbeFailure =
(probe: "lsof" | "windows-listeners") => (error: ProcessRunner.ProcessRunError) =>
Effect.logDebug("preview port process probe failed; probing configured URLs only", {
Expand All @@ -623,28 +742,38 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
}
}
if (hostPlatform === "win32") {
if (terminalByProcessId.size === 0 && configuredUrls.length === 0) {
return { discovered: [], configured: new Map() } satisfies WebProbeSnapshot;
}
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],
const runNetstat = (protocol: "tcp" | "tcpv6") =>
processRunner.run({
command: "netstat.exe",
args: ["-ano", "-p", protocol],
Comment thread
mweinbach marked this conversation as resolved.
timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS),
maxOutputBytes: 1024 * 1024,
})
.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([], configuredUrls);
maxOutputBytes: WINDOWS_NETSTAT_MAX_OUTPUT_BYTES,
});
const listeners = yield* Effect.all([runNetstat("tcp"), runNetstat("tcpv6")], {
concurrency: 2,
}).pipe(
// One failed query leaves a partial snapshot that could hide another process sharing a
// port across IPv4/IPv6, so fall back to configured URLs instead.
Effect.map(([tcp4, tcp6]) =>
exitedWithError(tcp4) || exitedWithError(tcp6)
? null
: parseWindowsListenerOutput(`${tcp4.stdout}\n${tcp6.stdout}`, terminalByProcessId),
),
Effect.catchTags({
ProcessSpawnError: recoverWindowsProbeFailure,
ProcessStdinError: recoverWindowsProbeFailure,
ProcessOutputLimitError: recoverWindowsProbeFailure,
ProcessReadError: recoverWindowsProbeFailure,
ProcessTimeoutError: recoverWindowsProbeFailure,
}),
);
return yield* withWindowsProcessNames(
yield* probeWebServers(listeners ?? [], configuredUrls),
Comment thread
mweinbach marked this conversation as resolved.
);
}
const recoverLsofProbeFailure = recoverProcessProbeFailure("lsof");
if (yield* Ref.get(lsofMissingRef)) {
Expand Down
Loading