perf(server): use native Windows APIs for preview discovery

apps/server/src/preview/PortScanner.test.ts

@@ -24,6 +24,7 @@ import { FetchHttpClient } from "effect/http";
 
 import * as ProcessRunner from "../processRunner.ts";
 import * as PortScanner from "./PortScanner.ts";
+import * as WindowsTcpListeners from "./WindowsTcpListeners.ts";
 
 vi.mock("@t3tools/provider-core/server/processGroup", { spy: true });
 
@@ -635,10 +636,19 @@ effectIt.effect("ignores a partial Windows netstat snapshot when one query exits
     requests.push(String(input));
     return Promise.resolve(new Response("hello", { headers: { "content-type": "text/html" } }));
   }) as typeof globalThis.fetch;
-  const layer = PortScanner.layer.pipe(
+  const layer = Layer.effect(PortScanner.PortDiscovery, PortScanner.make).pipe(
     Layer.provide(
       Layer.mergeAll(
         layerNoProc,
+        Layer.succeed(WindowsTcpListeners.WindowsTcpListeners, {
+          scan: Effect.fail(
+            new WindowsTcpListeners.WindowsTcpProbeError({
+              operation: "load",
+              cause: new Error("Native API unavailable"),
+            }),
+          ),
+          processName: () => Effect.succeed(null),
+        }),
         Layer.succeed(ProcessRunner.ProcessRunner, {
           run: (input) => {
             if (input.command === "tasklist.exe") {
@@ -648,16 +658,12 @@ effectIt.effect("ignores a partial Windows netstat snapshot when one query exits
             }
             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: "",
+              ...processOutput(
+                isIpv6
+                  ? ""
+                  : "  TCP    127.0.0.1:5173         0.0.0.0:0              LISTENING       1234\r\n",
+              ),
               code: isIpv6 && ipv6Fails ? ChildProcessSpawner.ExitCode(1) : null,
-              timedOut: false,
-              stdoutTruncated: false,
-              stderrTruncated: false,
-              stdoutInvalidUtf8: false,
-              stderrInvalidUtf8: false,
             });
           },
         }),
@@ -668,12 +674,7 @@ effectIt.effect("ignores a partial Windows netstat snapshot when one query exits
   );
 
   return Effect.gen(function* () {
-    const scanner = yield* PortScanner.PortDiscovery;
-    yield* scanner.registerTerminalProcesses({
-      threadId: "scanner-thread",
-      terminalId: "scanner-terminal",
-      processIds: [1234],
-    });
+    const scanner = yield* ownedScanner();
     expect(yield* scanner.scan()).toEqual([]);
     ipv6Fails = false;
     expect(yield* scanner.scan()).toMatchObject([
@@ -683,6 +684,80 @@ effectIt.effect("ignores a partial Windows netstat snapshot when one query exits
   }).pipe(Effect.provide(layer));
 });
 
+effectIt.effect.each([
+  ["native", "node"],
+  ["native error", "node"],
+  ["native null", "node"],
+  ["tasklist error", null],
+] as const)(
+  "filters native Windows ownership without probes and caches %s results",
+  ([lookup, expectedName]) => {
+    const requests: string[] = [];
+    const nameLookups: number[] = [];
+    const commands: string[] = [];
+    const layer = Layer.effect(PortScanner.PortDiscovery, PortScanner.make).pipe(
+      Layer.provide(
+        Layer.mergeAll(
+          layerNoProc,
+          Layer.succeed(HostProcess.Platform, "win32"),
+          Layer.succeed(ProcessRunner.ProcessRunner, {
+            run: (input) => {
+              commands.push(input.command);
+              return lookup === "tasklist error"
+                ? processProbeFailure(input)
+                : Effect.succeed(processOutput('"node.exe","1234","Console","1","50,000 K"\r\n'));
+            },
+          }),
+          Layer.succeed(WindowsTcpListeners.WindowsTcpListeners, {
+            scan: Effect.succeed([
+              { port: 5173, pid: 1234 },
+              { port: 5173, pid: 1234 },
+              { port: 5174, pid: 1234 },
+              { port: 5174, pid: 9999 },
+              { port: 5175, pid: 9999 },
+            ]),
+            processName: (pid) =>
+              Effect.suspend(() => {
+                nameLookups.push(pid);
+                if (lookup === "native") return Effect.succeed("node");
+                if (lookup === "native error") {
+                  return Effect.fail(
+                    new WindowsTcpListeners.WindowsTcpProbeError({
+                      operation: "process-name",
+                      cause: new Error("Native process name lookup unavailable"),
+                    }),
+                  );
+                }
+                return Effect.succeed(null);
+              }),
+          }),
+          FetchHttpClient.layer.pipe(
+            Layer.provide(
+              Layer.succeed(FetchHttpClient.Fetch, (input) => {
+                requests.push(String(input));
+                return Promise.resolve(new Response("app"));
+              }),
+            ),
+          ),
+        ),
+      ),
+    );
+    return Effect.gen(function* () {
+      const scanner = yield* ownedScanner();
+      const servers = yield* scanner.scan();
+      expect(servers).toMatchObject([{ port: 5173, pid: 1234, processName: expectedName }]);
+      expect(servers).toHaveLength(1);
+      expect(yield* scanner.scan()).toEqual(servers);
+      expect(requests).toEqual([]);
+      const configuredUrl = "http://localhost:5173/docs";
+      expect(yield* scanner.scan([configuredUrl])).toMatchObject([{ url: configuredUrl }]);
+      expect(requests).toEqual([configuredUrl]);
+      expect(nameLookups).toEqual([1234]);
+      expect(commands).toEqual(lookup === "native" ? [] : ["tasklist.exe"]);
+    }).pipe(Effect.provide(layer));
+  },
+);
+
 effectIt.effect("rechecks the socket owner before stopping a Linux server without lsof", () => {
   let holder = "4242";
   const missing = FileSystem.makeNoop({});

apps/server/src/preview/PortScanner.ts

@@ -24,8 +24,9 @@ import * as Scope from "effect/Scope";
 import * as Semaphore from "effect/Semaphore";
 import { FetchHttpClient, HttpClient } from "effect/http";
 
-import * as WindowsTcpListeners from "../process/windowsTcpListeners.ts";
+import * as WindowsTcpListenerQuery from "../process/windowsTcpListeners.ts";
 import * as ProcessRunner from "../processRunner.ts";
+import * as WindowsTcpListeners from "./WindowsTcpListeners.ts";
 
 export class PortDiscovery extends Context.Service<
   PortDiscovery,
@@ -298,7 +299,7 @@ const parseTasklistProcessName = (raw: string, pid: number): string | null => {
 };
 
 const discoveredWindowsListeners = (
-  listeners: ReadonlyArray<WindowsTcpListeners.WindowsTcpListener>,
+  listeners: ReadonlyArray<WindowsTcpListenerQuery.WindowsTcpListener>,
   owners: ReadonlyMap<number, ServerOwner>,
 ): ReadonlyArray<DiscoveredLocalServer> => {
   const seen = new Map<number, DiscoveredLocalServer>();
@@ -504,6 +505,7 @@ const isCommandNotFound = (error: ProcessRunner.ProcessSpawnError): boolean =>
 /** @public */
 export const make = Effect.gen(function* PortDiscoveryMake() {
   const processRunner = yield* ProcessRunner.ProcessRunner;
+  const windowsTcpListeners = yield* WindowsTcpListeners.WindowsTcpListeners;
   const fileSystem = yield* FileSystem.FileSystem;
   const hostPlatform = yield* HostProcess.Platform;
   const listenerDiscovery = (yield* HostProcess.Environment)[PORT_DISCOVERY_ENV] !== "0";
@@ -792,9 +794,27 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
     return { discovered: servers.filter(isOwned), configured } satisfies WebProbeSnapshot;
   });
 
+  const queryTasklistProcessName = Effect.fn("PortDiscovery.queryTasklistProcessName")(
+    (pid: number) =>
+      processRunner
+        .run({
+          command: "tasklist.exe",
+          args: ["/FO", "CSV", "/NH", "/FI", `PID eq ${pid}`],
+          timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS),
+          maxOutputBytes: 64 * 1024,
+        })
+        .pipe(
+          Effect.map((result) =>
+            exitedWithError(result) ? [] : [parseTasklistProcessName(result.stdout, pid)],
+          ),
+          Effect.orElseSucceed(() => []),
+        ),
+  );
+
   // tasklist costs 200-300ms per PID; query visible servers with bounded concurrency and cache names.
   const withWindowsProcessNames = Effect.fn("PortDiscovery.withWindowsProcessNames")(function* (
     snapshot: WebProbeSnapshot,
+    nameSource: "native" | "tasklist",
   ) {
     const servers = [...snapshot.discovered, ...snapshot.configured.values()];
     const pids = new Set(servers.flatMap((server) => (server.pid === null ? [] : [server.pid])));
@@ -807,26 +827,24 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
     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(
-            Effect.map((result) =>
-              exitedWithError(result)
-                ? []
-                : [
-                    [
-                      pid,
-                      { name: parseTasklistProcessName(result.stdout, pid), expiresAtMillis },
-                    ] as const,
-                  ],
-            ),
-            Effect.orElseSucceed(() => []),
-          ),
+        (nameSource === "native"
+          ? windowsTcpListeners.processName(pid).pipe(
+              Effect.flatMap((name) =>
+                name === null
+                  ? queryTasklistProcessName(pid).pipe(Effect.map((names) => [names[0] ?? null]))
+                  : Effect.succeed([name]),
+              ),
+              Effect.catchTags({
+                WindowsTcpProbeError: (error) =>
+                  Effect.logDebug("preview native process name lookup failed", {
+                    operation: error.operation,
+                  }).pipe(Effect.andThen(queryTasklistProcessName(pid))),
+              }),
+            )
+          : queryTasklistProcessName(pid)
+        ).pipe(
+          Effect.map((names) => names.map((name) => [pid, { name, expiresAtMillis }] as const)),
+        ),
       { concurrency: WINDOWS_PROCESS_NAME_CONCURRENCY },
     );
     const names = new Map([...cached, ...fetched.flat()]);
@@ -872,8 +890,23 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
     const resolveOwners = (pids: ReadonlyArray<number>) =>
       agentsPossible ? withAgentOwners(terminalOwners, pids) : Effect.succeed(terminalOwners);
     if (hostPlatform === "win32") {
+      const nativeListeners = yield* windowsTcpListeners.scan.pipe(
+        Effect.map((rows) => discoveredWindowsListeners(rows, terminalOwners)),
+        Effect.catchTags({
+          WindowsTcpProbeError: (error) =>
+            Effect.logDebug("preview native listener scan failed; falling back to netstat", {
+              operation: error.operation,
+            }).pipe(Effect.as(null)),
+        }),
+      );
+      if (nativeListeners !== null) {
+        return yield* withWindowsProcessNames(
+          yield* probeWebServers(nativeListeners, configuredUrls),
+          "native",
+        );
+      }
       const recoverWindowsProbeFailure = recoverProcessProbeFailure("windows-listeners");
-      const listeners = yield* WindowsTcpListeners.queryWindowsTcpListeners(
+      const listeners = yield* WindowsTcpListenerQuery.queryWindowsTcpListeners(
         processRunner,
         "netstat.exe",
       ).pipe(
@@ -890,6 +923,7 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
       );
       return yield* withWindowsProcessNames(
         yield* probeWebServers(listeners ?? [], configuredUrls),
+        "tasklist",
       );
     }
     const fromProc = yield* scanProcListeners(resolveOwners);
@@ -1228,7 +1262,9 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
   });
 }).pipe(Effect.withSpan("PortDiscovery.make"));
 
-export const layer = Layer.effect(PortDiscovery, make);
+export const layer = Layer.effect(PortDiscovery, make).pipe(
+  Layer.provide(WindowsTcpListeners.layer),
+);
 
 /** A scanner that tracks nothing, for tests that build provider runners without a host. */
 export const layerTest = Layer.mock(PortDiscovery)({

apps/server/src/preview/WindowsTcpListeners.test.ts

@@ -0,0 +1,52 @@
+import * as NodeNet from "node:net";
+
+import { it } from "@effect/vitest";
+import * as HostProcess from "@t3tools/shared/HostProcess";
+import { fileBasename } from "@t3tools/shared/path";
+import * as Effect from "effect/Effect";
+import { expect } from "vite-plus/test";
+
+import * as WindowsTcpListeners from "./WindowsTcpListeners.ts";
+
+it.effect("reads native Windows IPv4/IPv6 wildcard and loopback listeners and process names", () =>
+  Effect.gen(function* () {
+    if ((yield* HostProcess.Platform) !== "win32") return;
+    const ports: number[] = [];
+    for (const host of ["127.0.0.1", "0.0.0.0", "::1", "::"]) {
+      const server = yield* Effect.acquireRelease(
+        Effect.sync(() => NodeNet.createServer()),
+        (server) =>
+          Effect.promise(() => new Promise<void>((resolve) => server.close(() => resolve()))),
+      );
+      yield* Effect.promise(
+        () =>
+          new Promise<void>((resolve, reject) => {
+            server.once("error", reject);
+            server.listen({ host, port: 0, ipv6Only: true }, resolve);
+          }),
+      );
+      const address = server.address();
+      if (address === null || typeof address === "string") throw new Error("Missing TCP address");
+      ports.push(address.port);
+    }
+    const native = yield* WindowsTcpListeners.WindowsTcpListeners;
+    const rows = yield* native.scan;
+    for (const port of ports) expect(rows).toContainEqual({ port, pid: process.pid });
+    expect(yield* native.processName(process.pid)).toBe(
+      fileBasename(process.execPath).replace(/\.exe$/i, ""),
+    );
+    expect(yield* native.processName(0)).toBeNull();
+  }).pipe(Effect.provide(WindowsTcpListeners.layer)),
+);
+
+it.effect("rejects native TCP discovery on unsupported architectures", () =>
+  Effect.gen(function* () {
+    const native = yield* WindowsTcpListeners.WindowsTcpListeners;
+    const error = yield* Effect.flip(native.scan);
+    expect(error.operation).toBe("load");
+    expect(String(error.cause)).toContain("No native TCP discovery on ia32");
+  }).pipe(
+    Effect.provide(WindowsTcpListeners.layer),
+    Effect.provideService(HostProcess.Architecture, "ia32"),
+  ),
+);