1// @effect-diagnostics nodeBuiltinImport:off -- the stop test drives real server processes.
2import * as NodeChildProcess from "node:child_process";
3
4import { it as effectIt } from "@effect/vitest";
5import * as NodeServices from "@effect/platform-node/NodeServices";
6import {
7 CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS,
8 PREVIEW_URL_MAX_LENGTH,
9 type DiscoveredLocalServer,
10} from "@t3tools/contracts";
11import * as ProcessGroup from "@t3tools/provider-core/server/processGroup";
12import * as HostProcess from "@t3tools/shared/HostProcess";
13import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
14import * as Duration from "effect/Duration";
15import * as Effect from "effect/Effect";
16import * as Exit from "effect/Exit";
17import * as FileSystem from "effect/FileSystem";
18import * as Layer from "effect/Layer";
19import * as PlatformError from "effect/PlatformError";
20import * as Scope from "effect/Scope";
21import * as TestClock from "effect/testing/TestClock";
22import { expect, vi } from "vite-plus/test";
23import { FetchHttpClient } from "effect/http";
24
25import * as ProcessRunner from "../processRunner.ts";
26import * as PortScanner from "./PortScanner.ts";
27
28vi.mock("@t3tools/provider-core/server/processGroup", { spy: true });
29
30const processProbeFailure: ProcessRunner.ProcessRunner["Service"]["run"] = (input) =>
31 Effect.fail(
32 new ProcessRunner.ProcessSpawnError({
33 command: input.command,
34 argumentCount: input.args.length,
35 cwd: input.cwd,
36 cause: PlatformError.systemError({
37 _tag: "NotFound",
38 module: "ChildProcess",
39 method: "spawn",
40 description: "PowerShell is not installed in the test environment",
41 }),
42 }),
43 );
44
45const layerNoProc = FileSystem.layerNoop({});
46
47const layerProbeFailure = (
48 run: ProcessRunner.ProcessRunner["Service"]["run"],
49 fetch: typeof globalThis.fetch = globalThis.fetch,
50 fileSystem: Layer.Layer<FileSystem.FileSystem> = layerNoProc,
51) =>
52 PortScanner.layer.pipe(
53 Layer.provide(
54 Layer.mergeAll(
55 fileSystem,
56 Layer.succeed(ProcessRunner.ProcessRunner, { run }),
57 Layer.succeed(HostProcess.Platform, "linux"),
58 FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetch))),
59 ),
60 ),
61 );
62
63const LSOF_TEST_PORT = 43_123;
64
65const processOutput = (stdout: string): ProcessRunner.ProcessRunOutput => ({
66 stdout,
67 stderr: "",
68 code: ChildProcessSpawner.ExitCode(0),
69 timedOut: false,
70 stdoutTruncated: false,
71 stderrTruncated: false,
72 stdoutInvalidUtf8: false,
73 stderrInvalidUtf8: false,
74});
75
76const ownedScanner = (processIds: ReadonlyArray<number> = [1234]) =>
77 Effect.tap(PortScanner.PortDiscovery, (scanner) =>
78 scanner.registerTerminalProcesses({
79 threadId: "scanner-thread",
80 terminalId: "scanner-terminal",
81 processIds,
82 }),
83 );
84
85const layerLsofScanner = (input: {
86 readonly pid: () => number;
87 readonly fetch: typeof globalThis.fetch;
88 readonly stdout?: () => string;
89}) =>
90 PortScanner.layer.pipe(
91 Layer.provide(
92 Layer.mergeAll(
93 layerNoProc,
94 Layer.succeed(ProcessRunner.ProcessRunner, {
95 run: () =>
96 Effect.succeed({
97 stdout: input.stdout?.() ?? `p${input.pid()}\ncnode\nn*:${LSOF_TEST_PORT}\n`,
98 stderr: "",
99 code: null,
100 timedOut: false,
101 stdoutTruncated: false,
102 stderrTruncated: false,
103 stdoutInvalidUtf8: false,
104 stderrInvalidUtf8: false,
105 }),
106 }),
107 Layer.succeed(HostProcess.Platform, "linux"),
108 FetchHttpClient.layer.pipe(
109 Layer.provide(Layer.succeed(FetchHttpClient.Fetch, input.fetch)),
110 ),
111 ),
112 ),
113 );
114
115effectIt.effect("keeps a full configured URL when the discovered server root fails", () => {
116 const requests: string[] = [];
117 const configuredUrl = `http://localhost:${LSOF_TEST_PORT}/docs`;
118 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
119 const url = String(input);
120 requests.push(url);
121 return Promise.resolve(
122 url === configuredUrl
123 ? new Response("docs", { headers: { "content-type": "text/html" } })
124 : new Response("not found", {
125 status: 404,
126 headers: { "content-type": "text/html" },
127 }),
128 );
129 }) as typeof globalThis.fetch;
130 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
131
132 return Effect.gen(function* () {
133 const scanner = yield* PortScanner.PortDiscovery;
134 const servers = yield* scanner.scan([configuredUrl]);
135 expect(servers).toHaveLength(1);
136 expect(servers[0]?.url).toBe(configuredUrl);
137 expect(requests).toContain(configuredUrl);
138 }).pipe(Effect.provide(layer));
139});
140
141effectIt.effect("probes configured custom ports through a canonical loopback host", () => {
142 const customPort = 43_124;
143 const configuredUrl = `http://0.0.0.0:${customPort}/docs`;
144 const expectedUrl = `http://localhost:${customPort}/docs`;
145 const requests: string[] = [];
146 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
147 requests.push(String(input));
148 return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } }));
149 }) as typeof globalThis.fetch;
150 const layer = layerProbeFailure(processProbeFailure, fetchFn);
151
152 return Effect.gen(function* () {
153 const scanner = yield* PortScanner.PortDiscovery;
154 const servers = yield* scanner.scan([configuredUrl]);
155 expect(servers).toHaveLength(1);
156 expect(servers[0]?.host).toBe("localhost");
157 expect(servers[0]?.port).toBe(customPort);
158 expect(servers[0]?.url).toBe(expectedUrl);
159 expect(requests).toEqual([expectedUrl]);
160 }).pipe(Effect.provide(layer));
161});
162
163effectIt.effect("preserves explicit loopback hosts and bounds wildcard rewrites", () => {
164 const ipv4Url = "https://127.0.0.1:43125/docs";
165 const ipv6Url = "http://[::1]:43126/docs";
166 const wildcardPrefix = "http://0.0.0.0/";
167 const maximumWildcardUrl = `${wildcardPrefix}${"a".repeat(
168 PREVIEW_URL_MAX_LENGTH - wildcardPrefix.length,
169 )}`;
170 const requests: string[] = [];
171 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
172 requests.push(String(input));
173 return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } }));
174 }) as typeof globalThis.fetch;
175 const layer = layerProbeFailure(processProbeFailure, fetchFn);
176
177 return Effect.gen(function* () {
178 const scanner = yield* PortScanner.PortDiscovery;
179 const servers = yield* scanner.scan([ipv4Url, ipv6Url, maximumWildcardUrl]);
180 expect(servers.map((server) => server.url)).toEqual([ipv4Url, ipv6Url]);
181 expect(requests).toEqual([ipv4Url, ipv6Url]);
182 }).pipe(Effect.provide(layer));
183});
184
185effectIt.effect("projects configured paths independently for simultaneous subscribers", () => {
186 const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`;
187 const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`;
188 let ready = true;
189 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
190 const url = String(input);
191 return Promise.resolve(
192 ready && (url === docsUrl || url === adminUrl)
193 ? new Response("app", { headers: { "content-type": "text/html" } })
194 : new Response("not found", { status: 404 }),
195 );
196 }) as typeof globalThis.fetch;
197 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
198
199 return Effect.gen(function* () {
200 const scanner = yield* ownedScanner();
201 const docsSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = [];
202 const adminSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = [];
203 yield* scanner.subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, (servers) =>
204 Effect.sync(() => docsSnapshots.push(servers)),
205 );
206 yield* scanner.subscribe({ configuredUrls: [adminUrl], initialSnapshot: [] }, (servers) =>
207 Effect.sync(() => adminSnapshots.push(servers)),
208 );
209 yield* scanner.retain;
210
211 expect(docsSnapshots.at(-1)?.[0]?.url).toBe(docsUrl);
212 expect(adminSnapshots.at(-1)?.[0]?.url).toBe(adminUrl);
213
214 ready = false;
215 yield* TestClock.adjust(Duration.seconds(15));
216 expect(docsSnapshots.at(-1)?.[0]?.url).toBe(docsUrl);
217 expect(adminSnapshots.at(-1)?.[0]?.url).toBe(adminUrl);
218 }).pipe(Effect.scoped, Effect.provide(layer));
219});
220
221effectIt.effect(
222 "keeps each subscriber's candidates when their combined union exceeds the per-client cap",
223 () => {
224 const firstSubscriberUrls = Array.from(
225 { length: CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS },
226 (_, index) => `http://localhost:${LSOF_TEST_PORT}/app-${index}`,
227 );
228 const secondSubscriberUrl = `http://localhost:${LSOF_TEST_PORT}/app-${CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS}`;
229 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) =>
230 Promise.resolve(
231 String(input) === secondSubscriberUrl
232 ? new Response("app", { headers: { "content-type": "text/html" } })
233 : new Response("not found", { status: 404 }),
234 )) as typeof globalThis.fetch;
235 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
236
237 return Effect.gen(function* () {
238 const scanner = yield* PortScanner.PortDiscovery;
239 const secondSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = [];
240 yield* scanner.subscribe(
241 { configuredUrls: firstSubscriberUrls, initialSnapshot: [] },
242 () => Effect.void,
243 );
244 yield* scanner.subscribe(
245 { configuredUrls: [secondSubscriberUrl], initialSnapshot: [] },
246 (servers) => Effect.sync(() => secondSnapshots.push(servers)),
247 );
248 yield* scanner.retain;
249
250 expect(secondSnapshots.at(-1)?.[0]?.url).toBe(secondSubscriberUrl);
251 }).pipe(Effect.scoped, Effect.provide(layer));
252 },
253);
254
255effectIt.effect("stops probing a subscriber's configured paths after its scope closes", () => {
256 const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`;
257 const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`;
258 const requests: string[] = [];
259 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
260 const url = String(input);
261 requests.push(url);
262 return Promise.resolve(
263 url === docsUrl || url === adminUrl
264 ? new Response("app", { headers: { "content-type": "text/html" } })
265 : new Response("not found", { status: 404 }),
266 );
267 }) as typeof globalThis.fetch;
268 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
269
270 return Effect.gen(function* () {
271 const scanner = yield* PortScanner.PortDiscovery;
272 const docsScope = yield* Scope.make();
273 yield* scanner
274 .subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, () => Effect.void)
275 .pipe(Effect.provideService(Scope.Scope, docsScope));
276 yield* scanner.subscribe(
277 { configuredUrls: [adminUrl], initialSnapshot: [] },
278 () => Effect.void,
279 );
280 yield* scanner.retain;
281 yield* Scope.close(docsScope, Exit.void);
282
283 requests.length = 0;
284 yield* TestClock.adjust(Duration.seconds(15));
285 expect(requests).toContain(adminUrl);
286 expect(requests).not.toContain(docsUrl);
287 }).pipe(Effect.scoped, Effect.provide(layer));
288});
289
290effectIt.effect("uses the current configured fragment when readiness comes from cache", () => {
291 const requests: string[] = [];
292 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
293 requests.push(String(input));
294 return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } }));
295 }) as typeof globalThis.fetch;
296 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
297 const oldUrl = `http://localhost:${LSOF_TEST_PORT}/docs#old`;
298 const newUrl = `http://localhost:${LSOF_TEST_PORT}/docs#new`;
299
300 return Effect.gen(function* () {
301 const scanner = yield* PortScanner.PortDiscovery;
302 expect((yield* scanner.scan([oldUrl]))[0]?.url).toBe(oldUrl);
303 const requestCount = requests.length;
304 expect((yield* scanner.scan([newUrl]))[0]?.url).toBe(newUrl);
305 expect(requests).toHaveLength(requestCount);
306 }).pipe(Effect.provide(layer));
307});
308
309effectIt.effect(
310 "probes only a configured HTTPS URL, keeps it for an owned listener when the check fails, and does not follow redirects",
311 () => {
312 const configuredUrl = `https://localhost:${LSOF_TEST_PORT}/app`;
313 const requests: Array<readonly [string, string | undefined]> = [];
314 let respond = async () =>
315 new Response(null, { status: 302, headers: { location: "https://example.com" } });
316 const fetchFn = (async (
317 input: Parameters<typeof globalThis.fetch>[0],
318 init?: Parameters<typeof globalThis.fetch>[1],
319 ) => {
320 requests.push([String(input), init?.redirect]);
321 return respond();
322 }) as typeof globalThis.fetch;
323 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
324
325 return Effect.gen(function* () {
326 const scanner = yield* ownedScanner();
327 expect(yield* scanner.scan([configuredUrl])).toMatchObject([
328 { pid: 1234, url: configuredUrl },
329 ]);
330
331 respond = async () => new Response("unavailable", { status: 503 });
332 yield* TestClock.adjust(Duration.seconds(15));
333 expect(yield* scanner.scan([configuredUrl])).toMatchObject([
334 { pid: 1234, url: configuredUrl },
335 ]);
336
337 respond = () => Promise.reject(new TypeError("fetch failed"));
338 yield* TestClock.adjust(Duration.seconds(15));
339 expect(yield* scanner.scan([configuredUrl])).toMatchObject([
340 { pid: 1234, url: configuredUrl },
341 ]);
342 expect(requests).toEqual(Array.from({ length: 3 }, () => [configuredUrl, "manual"]));
343 }).pipe(Effect.provide(layer));
344 },
345);
346
347effectIt.effect("discovers owned listeners without sending HTTP or TLS probes", () => {
348 const owned = `p1234\ncnode\nn[::1]:${LSOF_TEST_PORT}\n`;
349 const unowned = `p5678\ncrpc\nn127.0.0.1:${LSOF_TEST_PORT}\n`;
350 let stdout = unowned;
351 const requests: string[] = [];
352 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
353 requests.push(String(input));
354 return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } }));
355 }) as typeof globalThis.fetch;
356 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn, stdout: () => stdout });
357
358 return Effect.gen(function* () {
359 const scanner = yield* ownedScanner();
360 expect(yield* scanner.scan()).toHaveLength(0);
361 stdout = `${owned}${unowned}`;
362 expect(yield* scanner.scan()).toHaveLength(0);
363 expect(requests).toEqual([]);
364
365 stdout = owned;
366 expect(yield* scanner.scan()).toHaveLength(1);
367 expect(requests).toEqual([]);
368
369 yield* scanner.unregisterTerminal({
370 threadId: "scanner-thread",
371 terminalId: "scanner-terminal",
372 });
373 expect(yield* scanner.scan()).toHaveLength(0);
374 expect(requests).toEqual([]);
375 }).pipe(Effect.provide(layer));
376});
377
378effectIt.effect(
379 "excludes configured URLs answering with HTTP errors, non-navigation responses, or non-documents",
380 () => {
381 let pid = 1;
382 let makeResponse = () =>
383 new Response("not found", { status: 404, headers: { "content-type": "text/html" } });
384 const fetchFn = ((_input: Parameters<typeof globalThis.fetch>[0]) =>
385 Promise.resolve(makeResponse())) as typeof globalThis.fetch;
386 const layer = layerLsofScanner({ pid: () => pid, fetch: fetchFn });
387 const configuredUrl = `http://localhost:${LSOF_TEST_PORT}/`;
388
389 return Effect.gen(function* () {
390 const scanner = yield* PortScanner.PortDiscovery;
391 const scan = () => scanner.scan([configuredUrl]);
392 expect(yield* scan()).toHaveLength(0);
393
394 pid += 1;
395 makeResponse = () =>
396 new Response("{}", { status: 200, headers: { "content-type": "application/json" } });
397 expect(yield* scan()).toHaveLength(0);
398
399 pid += 1;
400 makeResponse = () =>
401 new Response("ready", { status: 200, headers: { "content-type": "text/plain" } });
402 expect(yield* scan()).toHaveLength(0);
403
404 pid += 1;
405 makeResponse = () => new Response(null, { status: 304, headers: { location: "/cached" } });
406 expect(yield* scan()).toHaveLength(0);
407
408 pid += 1;
409 makeResponse = () =>
410 new Response(null, { status: 204, headers: { "content-type": "text/html" } });
411 expect(yield* scan()).toHaveLength(0);
412
413 pid += 1;
414 makeResponse = () => new Response(null, { status: 302 });
415 expect(yield* scan()).toHaveLength(0);
416
417 pid += 1;
418 makeResponse = () =>
419 new Response("<html />", {
420 status: 200,
421 headers: { "content-type": "application/xhtml+xml; charset=utf-8" },
422 });
423 expect(yield* scan()).toHaveLength(1);
424 }).pipe(Effect.provide(layer));
425 },
426);
427
428const PROC_NET_TCP = ` sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode
429 0: 0100007F:223D 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 5001 1 0000000000000000 100 0 0 10 0
430 1: 00000000:0016 00000000:0000 0A 00000000:00000000 00:00000000 00000000 0 0 5002 1 0000000000000000 100 0 0 10 0
431 2: 0100007F:223D 0100007F:C350 01 00000000:00000000 00:00000000 00000000 1000 0 5003 1 0000000000000000 20 4 30 10 -1
432 3: 0A00000F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 5005 1 0000000000000000 100 0 0 10 0
433`;
434const PROC_NET_TCP6 = ` sl local_address remote_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode
435 0: 00000000000000000000000000000000:0BB9 00000000000000000000000000000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 5004 1 0000000000000000 100 0 0 10 0
436`;
437
438effectIt.effect(
439 "prefers Linux /proc over installed lsof, and bounds repeated socket owner lookups",
440 () => {
441 let lsofSpawns = 0;
442 const fdWalks: string[] = [];
443 const procFiles: Record<string, string> = {
444 "/proc/net/tcp":
445 PROC_NET_TCP +
446 PROC_NET_TCP.split("\n")[1]!
447 .replace("0100007F:223D", "0101007F:1388")
448 .replace("5001", "5007"),
449 "/proc/net/tcp6": PROC_NET_TCP6,
450 "/proc/4242/comm": "python3\n",
451 };
452 const procFds: Record<string, Record<string, string>> = {
453 "4242": {
454 "0": "/dev/null",
455 "3": "socket:[5001]",
456 "4": "socket:[5004]",
457 "5": "socket:[5007]",
458 },
459 "77": { "5": "socket:[9999]" },
460 };
461 const missing = FileSystem.makeNoop({});
462 const fileSystem = FileSystem.layerNoop({
463 readFileString: (path) => {
464 const content = procFiles[path];
465 return content === undefined ? missing.readFileString(path) : Effect.succeed(content);
466 },
467 readDirectory: (path) => {
468 if (path === "/proc") return Effect.succeed(["self", "77", "4242"]);
469 const pid = /^\/proc\/(\d+)\/fd$/.exec(path)?.[1];
470 const fds = pid === undefined ? undefined : procFds[pid];
471 if (pid === undefined || fds === undefined) return missing.readDirectory(path);
472 fdWalks.push(pid);
473 return Effect.succeed(Object.keys(fds));
474 },
475 readLink: (path) => {
476 const [, pid, fd] = /^\/proc\/(\d+)\/fd\/(\d+)$/.exec(path) ?? [];
477 const target = pid && fd ? procFds[pid]?.[fd] : undefined;
478 return target === undefined ? missing.readLink(path) : Effect.succeed(target);
479 },
480 });
481 const requests: string[] = [];
482 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
483 requests.push(String(input));
484 return Promise.resolve(new Response("hello", { headers: { "content-type": "text/html" } }));
485 }) as typeof globalThis.fetch;
486 const layer = layerProbeFailure(
487 () => {
488 lsofSpawns += 1;
489 return Effect.succeed(processOutput("p4242\ncpython3\nn*:3001\nn*:8765\n"));
490 },
491 fetchFn,
492 fileSystem,
493 );
494
495 return Effect.gen(function* () {
496 const scanner = yield* ownedScanner([4242]);
497 const first = yield* scanner.scan();
498 expect(
499 first.map(({ port, pid, processName, url }) => ({ port, pid, processName, url })),
500 ).toEqual([
501 { port: 3001, pid: 4242, processName: "python3", url: "http://localhost:3001" },
502 { port: 8765, pid: 4242, processName: "python3", url: "http://localhost:8765" },
503 ]);
504 yield* scanner.scan();
505 expect(lsofSpawns).toBe(0);
506 expect(fdWalks).toEqual(["77", "4242", "77", "4242"]);
507 yield* scanner.scan();
508 yield* scanner.scan();
509 expect(fdWalks).toEqual(["77", "4242", "77", "4242", "77", "4242"]);
510 procFds["77"]!["6"] = "socket:[5006]";
511 procFiles["/proc/net/tcp6"] =
512 PROC_NET_TCP6 + PROC_NET_TCP.split("\n")[1]!.replace("5001", "5006");
513 expect((yield* scanner.scan()).map(({ port }) => port)).toEqual([3001]);
514 const walksAfterCollision = fdWalks.length;
515 procFiles["/proc/net/tcp"] = PROC_NET_TCP.split("\n")[0]!;
516 procFiles["/proc/net/tcp6"] = PROC_NET_TCP6.split("\n")[0]!;
517 expect(yield* scanner.scan()).toEqual([]);
518 expect(lsofSpawns).toBe(0);
519 expect(fdWalks).toHaveLength(walksAfterCollision);
520 expect(requests).toEqual([]);
521
522 procFiles["/proc/net/tcp"] = PROC_NET_TCP;
523 procFiles["/proc/net/tcp6"] = PROC_NET_TCP6;
524 expect((yield* scanner.scan()).map(({ pid }) => pid)).toEqual([4242, 4242]);
525 yield* scanner.retain.pipe(Effect.scoped);
526 procFds["77"] = procFds["4242"]!;
527 delete procFds["4242"];
528 procFiles["/proc/77/comm"] = "python3\n";
529 yield* scanner.registerTerminalProcesses({
530 threadId: "scanner-thread",
531 terminalId: "scanner-terminal",
532 processIds: [77],
533 });
534 expect((yield* scanner.scan()).map(({ port, pid }) => ({ port, pid }))).toEqual([
535 { port: 3001, pid: 77 },
536 { port: 8765, pid: 77 },
537 ]);
538 }).pipe(Effect.provide(layer));
539 },
540);
541
542effectIt.effect.each([
543 { mode: "missing", fallback: true },
544 { mode: "denied", fallback: true },
545 { mode: "invalid", fallback: true },
546 { mode: "fd-unavailable", fallback: true },
547 { mode: "fd-budget", fallback: true },
548 { mode: "fd-budget-no-lsof", fallback: true },
549 { mode: "ipv6-denied", fallback: true },
550 { mode: "ipv6-absent", fallback: false },
551 { mode: "empty", fallback: false },
552 { mode: "darwin", fallback: true },
553] as const)("uses lsof only if Linux /proc is unusable: $mode", ({ mode, fallback }) => {
554 let lsofSpawns = 0;
555 let procWalks = 0;
556 const missing = FileSystem.makeNoop({});
557 const layer = PortScanner.layer.pipe(
558 Layer.provide(
559 Layer.mergeAll(
560 FileSystem.layerNoop({
561 readFileString: (path) => {
562 if (!path.startsWith("/proc/net/tcp") || mode === "missing") {
563 return missing.readFileString(path);
564 }
565 if (mode === "denied" || (mode === "ipv6-denied" && path.endsWith("tcp6"))) {
566 return Effect.fail(
567 PlatformError.systemError({
568 _tag: "PermissionDenied",
569 module: "FileSystem",
570 method: "readFileString",
571 }),
572 );
573 }
574 if (mode === "ipv6-absent" && path.endsWith("tcp6")) {
575 return missing.readFileString(path);
576 }
577 if (mode === "invalid") return Effect.succeed("unusable table");
578 return Effect.succeed(
579 mode === "fd-unavailable" || mode.startsWith("fd-budget")
580 ? PROC_NET_TCP
581 : PROC_NET_TCP.split("\n")[0]!,
582 );
583 },
584 readDirectory: (path) => {
585 if (!mode.startsWith("fd-budget")) return missing.readDirectory(path);
586 if (path === "/proc") procWalks += 1;
587 if (mode === "fd-budget-no-lsof") {
588 if (path === "/proc") return Effect.succeed(["1234", "9999"]);
589 if (path === "/proc/1234/fd") return Effect.succeed(["3"]);
590 }
591 return Effect.succeed(
592 path === "/proc"
593 ? ["1", "1234"]
594 : Array.from({ length: 50_000 }, (_, i) => String(i)),
595 );
596 },
597 readLink: (path) =>
598 mode === "fd-budget-no-lsof" && path === "/proc/1234/fd/3"
599 ? Effect.succeed("socket:[5001]")
600 : mode.startsWith("fd-budget")
601 ? Effect.succeed("/dev/null")
602 : missing.readLink(path),
603 }),
604 Layer.succeed(ProcessRunner.ProcessRunner, {
605 run: (input) => {
606 lsofSpawns += 1;
607 if (mode === "fd-budget-no-lsof") return processProbeFailure(input);
608 return Effect.succeed(processOutput(`p1234\ncnode\nn*:${LSOF_TEST_PORT}\n`));
609 },
610 }),
611 Layer.succeed(HostProcess.Platform, mode === "darwin" ? "darwin" : "linux"),
612 FetchHttpClient.layer,
613 ),
614 ),
615 );
616 return Effect.gen(function* () {
617 const scanner = yield* ownedScanner();
618 expect((yield* scanner.scan()).map(({ port }) => port)).toEqual(
619 mode === "fd-budget-no-lsof" ? [8765] : fallback ? [LSOF_TEST_PORT] : [],
620 );
621 if (mode === "fd-budget-no-lsof") {
622 for (let i = 0; i < 3; i += 1) {
623 expect((yield* scanner.scan()).map(({ port }) => port)).toEqual([8765]);
624 }
625 expect(procWalks).toBe(4);
626 }
627 expect(lsofSpawns).toBe(fallback ? 1 : 0);
628 }).pipe(Effect.provide(layer));
629});
630
631effectIt.effect("ignores a partial Windows netstat snapshot when one query exits nonzero", () => {
632 let ipv6Fails = true;
633 const requests: string[] = [];
634 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
635 requests.push(String(input));
636 return Promise.resolve(new Response("hello", { headers: { "content-type": "text/html" } }));
637 }) as typeof globalThis.fetch;
638 const layer = PortScanner.layer.pipe(
639 Layer.provide(
640 Layer.mergeAll(
641 layerNoProc,
642 Layer.succeed(ProcessRunner.ProcessRunner, {
643 run: (input) => {
644 if (input.command === "tasklist.exe") {
645 return Effect.succeed(
646 processOutput('"node.exe","1234","Console","1","50,000 K"\r\n'),
647 );
648 }
649 const isIpv6 = input.args.includes("tcpv6");
650 return Effect.succeed({
651 stdout: isIpv6
652 ? ""
653 : " TCP 127.0.0.1:5173 0.0.0.0:0 LISTENING 1234\r\n",
654 stderr: "",
655 code: isIpv6 && ipv6Fails ? ChildProcessSpawner.ExitCode(1) : null,
656 timedOut: false,
657 stdoutTruncated: false,
658 stderrTruncated: false,
659 stdoutInvalidUtf8: false,
660 stderrInvalidUtf8: false,
661 });
662 },
663 }),
664 Layer.succeed(HostProcess.Platform, "win32"),
665 FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetchFn))),
666 ),
667 ),
668 );
669
670 return Effect.gen(function* () {
671 const scanner = yield* PortScanner.PortDiscovery;
672 yield* scanner.registerTerminalProcesses({
673 threadId: "scanner-thread",
674 terminalId: "scanner-terminal",
675 processIds: [1234],
676 });
677 expect(yield* scanner.scan()).toEqual([]);
678 ipv6Fails = false;
679 expect(yield* scanner.scan()).toMatchObject([
680 { port: 5173, pid: 1234, processName: "node", url: "http://localhost:5173" },
681 ]);
682 expect(requests).toEqual([]);
683 }).pipe(Effect.provide(layer));
684});
685
686effectIt.effect("rechecks the socket owner before stopping a Linux server without lsof", () => {
687 let holder = "4242";
688 const missing = FileSystem.makeNoop({});
689 const layer = layerProbeFailure(
690 (input) =>
691 input.command === "lsof"
692 ? processProbeFailure(input)
693 : Effect.succeed(processOutput("4242 1 4300\n5000 1 5300\n")),
694 () => Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } })),
695 FileSystem.layerNoop({
696 readFileString: (path) => {
697 if (path === "/proc/net/tcp") {
698 return Effect.succeed(PROC_NET_TCP.split("\n").slice(0, 2).join("\n"));
699 }
700 if (/^\/proc\/(4242|5000)\/comm$/.test(path)) return Effect.succeed("node\n");
701 if (/^\/proc\/(4242|5000)\/environ$/.test(path)) {
702 return Effect.succeed(`${PortScanner.AGENT_THREAD_ENV}=thread-agent\0`);
703 }
704 return missing.readFileString(path);
705 },
706 readDirectory: (path) => {
707 if (path === "/proc") return Effect.succeed(["4242", "5000"]);
708 if (/^\/proc\/(4242|5000)\/fd$/.test(path)) {
709 return Effect.succeed(path === `/proc/${holder}/fd` ? ["3"] : []);
710 }
711 return missing.readDirectory(path);
712 },
713 readLink: (path) =>
714 path === `/proc/${holder}/fd/3` ? Effect.succeed("socket:[5001]") : missing.readLink(path),
715 }),
716 );
717
718 return Effect.gen(function* () {
719 const signal = vi.spyOn(ProcessGroup, "signalProcessGroup").mockImplementation(() => {
720 throw Object.assign(new Error("No such process"), { code: "ESRCH" });
721 });
722 yield* Effect.addFinalizer(() => Effect.sync(() => signal.mockRestore()));
723 const scanner = yield* PortScanner.PortDiscovery;
724 expect(yield* scanner.scan()).toMatchObject([{ pid: 4242, port: 8765 }]);
725
726 // The socket outlives its original holder. Its old pid now belongs to another agent job.
727 holder = "5000";
728 const result = yield* scanner
729 .stopServer({ pid: 4242, port: 8765 })
730 .pipe(Effect.match({ onFailure: (error) => error.reason, onSuccess: () => null }));
731 expect(signal).not.toHaveBeenCalled();
732 expect(result).toBe("not-found");
733 expect(yield* scanner.scan()).toMatchObject([{ pid: 5000, port: 8765 }]);
734 yield* scanner.stopServer({ pid: 5000, port: 8765 });
735 expect(signal).toHaveBeenCalledExactlyOnceWith(5300, "SIGTERM");
736 }).pipe(Effect.provide(layer));
737});
738
739effectIt.effect.each(["linux", "darwin", "win32"] as const)(
740 "reads no listeners on %s while T3CODE_PORT_DISCOVERY=0, and still probes configured URLs",
741 (platform) => {
742 const configuredUrl = `http://localhost:${LSOF_TEST_PORT}/docs`;
743 const commands: string[] = [];
744 const fileReads: string[] = [];
745 const requests: string[] = [];
746 const missing = FileSystem.makeNoop({});
747 const layer = PortScanner.layer.pipe(
748 Layer.provide(
749 Layer.mergeAll(
750 FileSystem.layerNoop({
751 readFileString: (path) => {
752 fileReads.push(path);
753 return missing.readFileString(path);
754 },
755 readDirectory: (path) => {
756 fileReads.push(path);
757 return missing.readDirectory(path);
758 },
759 }),
760 Layer.succeed(ProcessRunner.ProcessRunner, {
761 run: (input) => {
762 commands.push(input.command);
763 return Effect.succeed(processOutput(`p1234\ncnode\nn*:${LSOF_TEST_PORT}\n`));
764 },
765 }),
766 Layer.succeed(HostProcess.Platform, platform),
767 Layer.succeed(HostProcess.Environment, { T3CODE_PORT_DISCOVERY: "0" }),
768 FetchHttpClient.layer.pipe(
769 Layer.provide(
770 Layer.succeed(FetchHttpClient.Fetch, (input) => {
771 requests.push(String(input));
772 return Promise.resolve(
773 new Response("docs", { headers: { "content-type": "text/html" } }),
774 );
775 }),
776 ),
777 ),
778 ),
779 ),
780 );
781
782 return Effect.gen(function* () {
783 const scanner = yield* ownedScanner();
784 scanner.trackAgent();
785 expect(yield* scanner.scan()).toEqual([]);
786 expect(yield* scanner.scan([configuredUrl])).toMatchObject([
787 { port: LSOF_TEST_PORT, pid: null, terminal: null, url: configuredUrl },
788 ]);
789 expect(requests).toEqual([configuredUrl]);
790 expect(commands).toEqual([]);
791 expect(fileReads).toEqual([]);
792 }).pipe(Effect.provide(layer));
793 },
794);
795
796effectIt.effect.each(["linux", "darwin"] as const)(
797 "attributes listeners to the agent thread in their environment on %s, after the agent exits",
798 (platform) => {
799 const agentPort = 43_124;
800 const otherPort = 43_125;
801 const agentListener = `p4242\ncnode\nn127.0.0.1:${agentPort}\n`;
802 const otherListener = `p5000\ncpostgres\nn127.0.0.1:${otherPort}\n`;
803 let lsofStdout = otherListener;
804 let lsofFails = false;
805 let environmentFails = false;
806 let lsofCalls = 0;
807 let environmentReads = 0;
808 const requests: string[] = [];
809 const missing = FileSystem.makeNoop({});
810 const procEnviron: Record<string, string> = {
811 "/proc/4242/environ": `PATH=/usr/bin\0${PortScanner.AGENT_THREAD_ENV}=thread-agent\0`,
812 "/proc/5000/environ": "PATH=/usr/bin\0",
813 };
814 const layer = PortScanner.layer.pipe(
815 Layer.provide(
816 Layer.mergeAll(
817 FileSystem.layerNoop({
818 readFileString: (path) => {
819 const content = procEnviron[path];
820 if (content === undefined) return missing.readFileString(path);
821 environmentReads += 1;
822 return Effect.succeed(content);
823 },
824 }),
825 Layer.succeed(ProcessRunner.ProcessRunner, {
826 run: (input) => {
827 if (input.command === "lsof") {
828 lsofCalls += 1;
829 return lsofFails
830 ? Effect.fail(
831 new ProcessRunner.ProcessTimeoutError({
832 command: input.command,
833 argumentCount: input.args.length,
834 cwd: input.cwd,
835 timeoutMs: 5_000,
836 }),
837 )
838 : Effect.succeed(processOutput(lsofStdout));
839 }
840 // A direct child of this process, so stop refuses it without signaling anything.
841 if (input.args[0] === "-A") {
842 return Effect.succeed(processOutput(`4242 ${process.pid} 4242\n`));
843 }
844 if (environmentFails) {
845 return Effect.fail(
846 new ProcessRunner.ProcessTimeoutError({
847 command: input.command,
848 argumentCount: input.args.length,
849 timeoutMs: 5_000,
850 }),
851 );
852 }
853 environmentReads += 1;
854 return Effect.succeed(
855 processOutput(
856 ` 4242 node server.js PATH=/usr/bin ${PortScanner.AGENT_THREAD_ENV}=thread-agent HOME=/Users/dev\n` +
857 " 5000 postgres -D /data PATH=/usr/bin\n",
858 ),
859 );
860 },
861 }),
862 Layer.succeed(HostProcess.Platform, platform),
863 FetchHttpClient.layer.pipe(
864 Layer.provide(
865 Layer.succeed(FetchHttpClient.Fetch, (input) => {
866 requests.push(String(input));
867 return Promise.resolve(
868 new Response("app", { headers: { "content-type": "text/html" } }),
869 );
870 }),
871 ),
872 ),
873 ),
874 ),
875 );
876
877 return Effect.gen(function* () {
878 const scanner = yield* PortScanner.PortDiscovery;
879 // Startup reads listeners once. With none marked, later scans stay idle.
880 expect(yield* scanner.scan()).toEqual([]);
881 expect(yield* scanner.scan()).toEqual([]);
882 expect(lsofCalls).toBe(1);
883
884 // The agent exits before any client looks, and the first read fails; its server is still
885 // found.
886 lsofStdout = `${agentListener}${otherListener}`;
887 scanner.trackAgent()();
888 lsofFails = true;
889 expect(yield* scanner.scan()).toEqual([]);
890 lsofFails = false;
891 if (platform === "darwin") {
892 environmentFails = true;
893 expect(yield* scanner.scan()).toEqual([]);
894 environmentFails = false;
895 }
896 expect(yield* scanner.scan()).toMatchObject([
897 { port: agentPort, pid: 4242, terminal: null, agent: { threadId: "thread-agent" } },
898 ]);
899 expect(requests).toEqual([]);
900
901 expect(yield* scanner.scan()).toHaveLength(1);
902 expect(environmentReads).toBe(2);
903
904 // Stop reads its target's environment again rather than trust the cache, and still finds it.
905 const stop = yield* Effect.flip(scanner.stopServer({ pid: 4242, port: agentPort }));
906 expect(stop.reason).toBe("shared-group");
907 expect(environmentReads).toBe(3);
908
909 lsofStdout = otherListener;
910 expect(yield* scanner.scan()).toEqual([]);
911 const lsofCallsAfterExit = lsofCalls;
912 expect(yield* scanner.scan()).toEqual([]);
913 expect(lsofCalls).toBe(lsofCallsAfterExit);
914 }).pipe(Effect.provide(layer));
915 },
916);
917
918effectIt.effect("stops only a job's own process group", () =>
919 Effect.sync(() => {
920 const serverPid = 100;
921 const rows = [
922 { pid: 10, ppid: 1, pgid: 10 }, // `vp run dev`, whose job holds this server
923 { pid: 20, ppid: 10, pgid: 10 }, // the web dev server in the same job
924 { pid: serverPid, ppid: 10, pgid: 10 },
925 { pid: 200, ppid: serverPid, pgid: 200 }, // agent CLI
926 { pid: 210, ppid: 200, pgid: 200 }, // a server it spawned without a group of its own
927 { pid: 300, ppid: 200, pgid: 300 }, // agent tool shell
928 { pid: 400, ppid: 300, pgid: 400 }, // `npm run dev &`
929 { pid: 401, ppid: 400, pgid: 400 },
930 { pid: 500, ppid: serverPid, pgid: 500 }, // terminal shell
931 { pid: 600, ppid: 500, pgid: 600 }, // terminal job
932 ];
933 const table = new Map(rows.map((row) => [row.pid, row]));
934 expect(PortScanner.stoppableProcessGroup(table, 401, serverPid)).toBe(400);
935 expect(PortScanner.stoppableProcessGroup(table, 600, serverPid)).toBe(600);
936 for (const pid of [20, 210, 500, 999]) {
937 expect(PortScanner.stoppableProcessGroup(table, pid, serverPid)).toBeNull();
938 }
939 }),
940);
941
942const AGENT_SERVER_SCRIPT = `
943const server = require("node:http").createServer((_, response) => {
944 response.writeHead(200, { "content-type": "text/html" });
945 response.end("<html></html>");
946});
947server.listen(0, "127.0.0.1", () => {
948 process.stdout.write(process.pid + " " + server.address().port + "\\n");
949});
950`;
951
952/**
953 * A real HTML server with an agent's environment, started the way an agent's tool shell starts a
954 * job: under a detached shell with job control, in a group of its own. Release kills only the pids
955 * it started.
956 */
957const agentServer = () =>
958 Effect.acquireRelease(
959 Effect.callback<{
960 readonly child: NodeChildProcess.ChildProcess;
961 readonly pid: number;
962 readonly port: number;
963 readonly exited: Promise<void>;
964 }>((resume) => {
965 const env = { ...process.env, [PortScanner.AGENT_THREAD_ENV]: "thread-live" };
966 // dash ignores `set -m` without a terminal; bash, like agent tool shells, honors it.
967 const child = NodeChildProcess.spawn(
968 "bash",
969 ["-c", 'set -m; "$0" -e "$1" & wait', process.execPath, AGENT_SERVER_SCRIPT],
970 { detached: true, env, stdio: ["ignore", "pipe", "ignore"] },
971 );
972 const exited = new Promise<void>((resolve) => child.once("exit", () => resolve()));
973 child.once("error", (error) => resume(Effect.die(error)));
974 child.stdout?.once("data", (chunk) => {
975 const [pid, port] = String(chunk).trim().split(" ").map(Number);
976 resume(Effect.succeed({ child, pid: pid!, port: port!, exited }));
977 });
978 }),
979 ({ child, pid }) =>
980 Effect.sync(() => {
981 for (const target of [pid, child.pid]) {
982 try {
983 if (target !== undefined) process.kill(target, "SIGKILL");
984 } catch {}
985 }
986 }),
987 );
988
989const isRunning = (pid: number): boolean => {
990 try {
991 process.kill(pid, 0);
992 return true;
993 } catch {
994 return false;
995 }
996};
997
998effectIt.live.skipIf(HostProcess.Platform.defaultValue() === "win32")(
999 "stops an agent's server through its job's group, and only a listed server",
1000 () =>
1001 Effect.gen(function* () {
1002 const scanner = yield* PortScanner.PortDiscovery;
1003 yield* Effect.acquireRelease(Effect.sync(scanner.trackAgent), (release) =>
1004 Effect.sync(release),
1005 );
1006 const job = yield* agentServer();
1007
1008 expect(yield* scanner.scan()).toEqual(
1009 expect.arrayContaining([
1010 expect.objectContaining({
1011 pid: job.pid,
1012 port: job.port,
1013 agent: { threadId: "thread-live" },
1014 }),
1015 ]),
1016 );
1017
1018 const unlisted = yield* Effect.flip(scanner.stopServer({ pid: process.pid, port: job.port }));
1019 expect(unlisted.reason).toBe("not-found");
1020
1021 yield* scanner.stopServer({ pid: job.pid, port: job.port });
1022 // The shell waits on its job, so it exits once the job has.
1023 yield* Effect.promise(() => job.exited);
1024 expect(isRunning(job.pid)).toBe(false);
1025 }).pipe(
1026 Effect.scoped,
1027 Effect.provide(
1028 PortScanner.layer.pipe(
1029 Layer.provide(ProcessRunner.layer),
1030 Layer.provide(Layer.mergeAll(NodeServices.layer, FetchHttpClient.layer)),
1031 ),
1032 ),
1033 ),
1034);