1import * as NodeNet from "node:net";
2
3import { it as effectIt } from "@effect/vitest";
4import {
5 CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS,
6 PREVIEW_URL_MAX_LENGTH,
7 type DiscoveredLocalServer,
8} from "@t3tools/contracts";
9import * as HostProcess from "@t3tools/shared/HostProcess";
10import * as Cause from "effect/Cause";
11import * as Clock from "effect/Clock";
12import * as Duration from "effect/Duration";
13import * as Effect from "effect/Effect";
14import * as Exit from "effect/Exit";
15import * as Fiber from "effect/Fiber";
16import * as FileSystem from "effect/FileSystem";
17import * as Layer from "effect/Layer";
18import * as PlatformError from "effect/PlatformError";
19import * as Scope from "effect/Scope";
20import * as TestClock from "effect/testing/TestClock";
21import * as Tracer from "effect/Tracer";
22import { expect } from "vite-plus/test";
23import { FetchHttpClient } from "effect/http";
24
25import * as ProcessRunner from "../processRunner.ts";
26import * as PortScanner from "./PortScanner.ts";
27const processProbeFailure: ProcessRunner.ProcessRunner["Service"]["run"] = (input) =>
28 Effect.fail(
29 new ProcessRunner.ProcessSpawnError({
30 command: input.command,
31 argumentCount: input.args.length,
32 cwd: input.cwd,
33 cause: PlatformError.systemError({
34 _tag: "NotFound",
35 module: "ChildProcess",
36 method: "spawn",
37 description: "PowerShell is not installed in the test environment",
38 }),
39 }),
40 );
41
42const layerTestProcessRunner = Layer.succeed(ProcessRunner.ProcessRunner, {
43 run: processProbeFailure,
44});
45
46/** A host without `/proc`, so a missing `lsof` leaves only configured URLs. */
47const layerNoProc = FileSystem.layerNoop({});
48
49const layerProbeFailure = (
50 run: ProcessRunner.ProcessRunner["Service"]["run"],
51 fetch: typeof globalThis.fetch = globalThis.fetch,
52 fileSystem: Layer.Layer<FileSystem.FileSystem> = layerNoProc,
53) =>
54 PortScanner.layer.pipe(
55 Layer.provide(
56 Layer.mergeAll(
57 fileSystem,
58 Layer.succeed(ProcessRunner.ProcessRunner, { run }),
59 Layer.succeed(HostProcess.Platform, "linux"),
60 FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetch))),
61 ),
62 ),
63 );
64
65const layerTestPortDiscovery = PortScanner.layer.pipe(
66 Layer.provide(
67 Layer.mergeAll(
68 layerNoProc,
69 layerTestProcessRunner,
70 Layer.succeed(HostProcess.Platform, "win32"),
71 FetchHttpClient.layer,
72 ),
73 ),
74);
75
76const LSOF_TEST_PORT = 43_123;
77
78/** The scanner with `processIds` registered to a T3 terminal, so their listeners get probes. */
79const ownedScanner = (processIds: ReadonlyArray<number> = [1234]) =>
80 Effect.tap(PortScanner.PortDiscovery, (scanner) =>
81 scanner.registerTerminalProcesses({
82 threadId: "scanner-thread",
83 terminalId: "scanner-terminal",
84 processIds,
85 }),
86 );
87
88const layerLsofScanner = (input: {
89 readonly pid: () => number;
90 readonly fetch: typeof globalThis.fetch;
91 readonly stdout?: () => string;
92}) =>
93 PortScanner.layer.pipe(
94 Layer.provide(
95 Layer.mergeAll(
96 layerNoProc,
97 Layer.succeed(ProcessRunner.ProcessRunner, {
98 run: () =>
99 Effect.succeed({
100 stdout: input.stdout?.() ?? `p${input.pid()}\ncnode\nn*:${LSOF_TEST_PORT}\n`,
101 stderr: "",
102 code: null,
103 timedOut: false,
104 stdoutTruncated: false,
105 stderrTruncated: false,
106 stdoutInvalidUtf8: false,
107 stderrInvalidUtf8: false,
108 }),
109 }),
110 Layer.succeed(HostProcess.Platform, "linux"),
111 FetchHttpClient.layer.pipe(
112 Layer.provide(Layer.succeed(FetchHttpClient.Fetch, input.fetch)),
113 ),
114 ),
115 ),
116 );
117
118const openServer = (
119 port: number,
120 onConnection: (socket: NodeNet.Socket) => void,
121): Effect.Effect<NodeNet.Server | null> =>
122 Effect.callback((resume) => {
123 const server = NodeNet.createServer(onConnection);
124 server.once("error", () => {
125 resume(Effect.succeed(null));
126 });
127 server.listen(port, "127.0.0.1", () => {
128 resume(Effect.succeed(server));
129 });
130 return Effect.sync(() => {
131 server.close();
132 });
133 });
134
135const closeServer = (server: NodeNet.Server): Effect.Effect<void> =>
136 Effect.callback((resume) => {
137 server.close(() => resume(Effect.void));
138 });
139
140const openCommonDevServer = Effect.fn("PortScannerTest.openCommonDevServer")(function* (
141 onConnection: (socket: NodeNet.Socket) => void,
142) {
143 const server = yield* openServer(0, onConnection);
144 const address = server?.address();
145 if (!server || !address || typeof address === "string") {
146 return yield* Effect.die(new Error("Could not open the preview scanner test listener"));
147 }
148 return { port: address.port, server };
149});
150
151const commonDevServer = Effect.acquireRelease(
152 openCommonDevServer((socket) => {
153 socket.once("data", () => {
154 socket.end("HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: 5\r\n\r\nhello");
155 });
156 }),
157 ({ server }) => closeServer(server),
158);
159
160const commonNonHttpServer = Effect.acquireRelease(
161 openCommonDevServer((socket) => {
162 socket.on("error", () => undefined);
163 socket.once("data", () => socket.end("MYSQL\r\n\r\n"));
164 }),
165 ({ server }) => closeServer(server),
166);
167
168/**
169 * Integration tests against a real TCP listener. We provide the Windows host
170 * platform with a failing listener probe so the tests exercise configured URLs
171 * without depending on `lsof` being installed.
172 */
173effectIt.layer(layerTestPortDiscovery)("PortDiscovery integration (configured URLs)", (it) => {
174 it.effect(
175 "scan() returns a configured HTTP server we just opened",
176 Effect.fn("PortScannerTest.scanFindsCommonDevServer")(function* () {
177 const { port } = yield* commonDevServer;
178 const scanner = yield* PortScanner.PortDiscovery;
179 const result = yield* scanner.scan([`http://localhost:${port}`]);
180 const found = result.find((server) => server.port === port);
181 expect(found).toBeDefined();
182 expect(found?.host).toBe("localhost");
183 }),
184 );
185
186 it.effect(
187 "scan() excludes a listening port that does not speak HTTP",
188 Effect.fn("PortScannerTest.scanExcludesNonHttpServer")(function* () {
189 const { port } = yield* commonNonHttpServer;
190 const scanner = yield* PortScanner.PortDiscovery;
191 const result = yield* scanner.scan([`http://localhost:${port}`]);
192 expect(result.some((server) => server.port === port)).toBe(false);
193 }),
194 );
195
196 it.effect(
197 "retain drives an immediate broadcast to subscribers",
198 Effect.fn("PortScannerTest.retainBroadcastsImmediately")(function* () {
199 const { port } = yield* commonDevServer;
200 const received: number[] = [];
201 const scanner = yield* PortScanner.PortDiscovery;
202 yield* scanner.subscribe(
203 { configuredUrls: [`http://localhost:${port}`], initialSnapshot: [] },
204 (servers) =>
205 Effect.sync(() => {
206 for (const server of servers) received.push(server.port);
207 }),
208 );
209 yield* scanner.retain;
210 expect(received).toContain(port);
211 }),
212 );
213});
214
215effectIt.effect("revalidates a successful HTML probe after its cache entry expires", () => {
216 let responds = true;
217 const requests: string[] = [];
218 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
219 requests.push(String(input));
220 return responds
221 ? Promise.resolve(new Response("hello", { headers: { "content-type": "text/html" } }))
222 : Promise.reject(new TypeError("not HTTP"));
223 }) as typeof globalThis.fetch;
224 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
225
226 return Effect.gen(function* () {
227 const scanner = yield* ownedScanner();
228 expect(yield* scanner.scan()).toHaveLength(1);
229 expect(yield* scanner.scan()).toHaveLength(1);
230 expect(requests).toEqual([`http://localhost:${LSOF_TEST_PORT}/`]);
231
232 responds = false;
233 yield* TestClock.adjust(Duration.seconds(15));
234 expect(yield* scanner.scan()).toHaveLength(0);
235 expect(requests).toEqual([
236 `http://localhost:${LSOF_TEST_PORT}/`,
237 `http://localhost:${LSOF_TEST_PORT}/`,
238 `https://localhost:${LSOF_TEST_PORT}/`,
239 ]);
240 }).pipe(Effect.provide(layer));
241});
242
243effectIt.effect("keeps a full configured URL when the discovered server root fails", () => {
244 const requests: string[] = [];
245 const configuredUrl = `http://localhost:${LSOF_TEST_PORT}/docs`;
246 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
247 const url = String(input);
248 requests.push(url);
249 return Promise.resolve(
250 url === configuredUrl
251 ? new Response("docs", { headers: { "content-type": "text/html" } })
252 : new Response("not found", {
253 status: 404,
254 headers: { "content-type": "text/html" },
255 }),
256 );
257 }) as typeof globalThis.fetch;
258 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
259
260 return Effect.gen(function* () {
261 const scanner = yield* PortScanner.PortDiscovery;
262 const servers = yield* scanner.scan([configuredUrl]);
263 expect(servers).toHaveLength(1);
264 expect(servers[0]?.url).toBe(configuredUrl);
265 expect(requests).toContain(configuredUrl);
266 }).pipe(Effect.provide(layer));
267});
268
269effectIt.effect("probes configured custom ports through a canonical loopback host", () => {
270 const customPort = 43_124;
271 const configuredUrl = `http://0.0.0.0:${customPort}/docs`;
272 const expectedUrl = `http://localhost:${customPort}/docs`;
273 const requests: string[] = [];
274 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
275 requests.push(String(input));
276 return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } }));
277 }) as typeof globalThis.fetch;
278 const layer = layerProbeFailure(processProbeFailure, fetchFn);
279
280 return Effect.gen(function* () {
281 const scanner = yield* PortScanner.PortDiscovery;
282 const servers = yield* scanner.scan([configuredUrl]);
283 expect(servers).toHaveLength(1);
284 expect(servers[0]?.host).toBe("localhost");
285 expect(servers[0]?.port).toBe(customPort);
286 expect(servers[0]?.url).toBe(expectedUrl);
287 expect(requests).toEqual([expectedUrl]);
288 }).pipe(Effect.provide(layer));
289});
290
291effectIt.effect("preserves explicit loopback hosts and bounds wildcard rewrites", () => {
292 const ipv4Url = "https://127.0.0.1:43125/docs";
293 const ipv6Url = "http://[::1]:43126/docs";
294 const wildcardPrefix = "http://0.0.0.0/";
295 const maximumWildcardUrl = `${wildcardPrefix}${"a".repeat(
296 PREVIEW_URL_MAX_LENGTH - wildcardPrefix.length,
297 )}`;
298 const requests: string[] = [];
299 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
300 requests.push(String(input));
301 return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } }));
302 }) as typeof globalThis.fetch;
303 const layer = layerProbeFailure(processProbeFailure, fetchFn);
304
305 return Effect.gen(function* () {
306 const scanner = yield* PortScanner.PortDiscovery;
307 const servers = yield* scanner.scan([ipv4Url, ipv6Url, maximumWildcardUrl]);
308 expect(servers.map((server) => server.url)).toEqual([ipv4Url, ipv6Url]);
309 expect(requests).toEqual([ipv4Url, ipv6Url]);
310 }).pipe(Effect.provide(layer));
311});
312
313effectIt.effect("projects configured paths independently for simultaneous subscribers", () => {
314 const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`;
315 const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`;
316 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
317 const url = String(input);
318 return Promise.resolve(
319 url === docsUrl || url === adminUrl
320 ? new Response("app", { headers: { "content-type": "text/html" } })
321 : new Response("not found", { status: 404 }),
322 );
323 }) as typeof globalThis.fetch;
324 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
325
326 return Effect.gen(function* () {
327 const scanner = yield* PortScanner.PortDiscovery;
328 const docsSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = [];
329 const adminSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = [];
330 yield* scanner.subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, (servers) =>
331 Effect.sync(() => docsSnapshots.push(servers)),
332 );
333 yield* scanner.subscribe({ configuredUrls: [adminUrl], initialSnapshot: [] }, (servers) =>
334 Effect.sync(() => adminSnapshots.push(servers)),
335 );
336 yield* scanner.retain;
337
338 expect(docsSnapshots.at(-1)?.[0]?.url).toBe(docsUrl);
339 expect(adminSnapshots.at(-1)?.[0]?.url).toBe(adminUrl);
340 }).pipe(Effect.scoped, Effect.provide(layer));
341});
342
343effectIt.effect(
344 "keeps each subscriber's candidates when their combined union exceeds the per-client cap",
345 () => {
346 const firstSubscriberUrls = Array.from(
347 { length: CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS },
348 (_, index) => `http://localhost:${LSOF_TEST_PORT}/app-${index}`,
349 );
350 const secondSubscriberUrl = `http://localhost:${LSOF_TEST_PORT}/app-${CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS}`;
351 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) =>
352 Promise.resolve(
353 String(input) === secondSubscriberUrl
354 ? new Response("app", { headers: { "content-type": "text/html" } })
355 : new Response("not found", { status: 404 }),
356 )) as typeof globalThis.fetch;
357 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
358
359 return Effect.gen(function* () {
360 const scanner = yield* PortScanner.PortDiscovery;
361 const secondSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = [];
362 yield* scanner.subscribe(
363 { configuredUrls: firstSubscriberUrls, initialSnapshot: [] },
364 () => Effect.void,
365 );
366 yield* scanner.subscribe(
367 { configuredUrls: [secondSubscriberUrl], initialSnapshot: [] },
368 (servers) => Effect.sync(() => secondSnapshots.push(servers)),
369 );
370 yield* scanner.retain;
371
372 expect(secondSnapshots.at(-1)?.[0]?.url).toBe(secondSubscriberUrl);
373 }).pipe(Effect.scoped, Effect.provide(layer));
374 },
375);
376
377effectIt.effect("stops probing a subscriber's configured paths after its scope closes", () => {
378 const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`;
379 const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`;
380 const requests: string[] = [];
381 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
382 const url = String(input);
383 requests.push(url);
384 return Promise.resolve(
385 url === docsUrl || url === adminUrl
386 ? new Response("app", { headers: { "content-type": "text/html" } })
387 : new Response("not found", { status: 404 }),
388 );
389 }) as typeof globalThis.fetch;
390 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
391
392 return Effect.gen(function* () {
393 const scanner = yield* PortScanner.PortDiscovery;
394 const docsScope = yield* Scope.make();
395 yield* scanner
396 .subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, () => Effect.void)
397 .pipe(Effect.provideService(Scope.Scope, docsScope));
398 yield* scanner.subscribe(
399 { configuredUrls: [adminUrl], initialSnapshot: [] },
400 () => Effect.void,
401 );
402 yield* scanner.retain;
403 yield* Scope.close(docsScope, Exit.void);
404
405 requests.length = 0;
406 yield* TestClock.adjust(Duration.seconds(15));
407 expect(requests).toContain(adminUrl);
408 expect(requests).not.toContain(docsUrl);
409 }).pipe(Effect.scoped, Effect.provide(layer));
410});
411
412effectIt.effect("writes no poll span while no client retains the scanner", () => {
413 let pollSpans = 0;
414 const tracer = Tracer.make({
415 span: (options) => {
416 if (options.name === "PortDiscovery.pollTick") pollSpans += 1;
417 return new Tracer.NativeSpan(options);
418 },
419 });
420 const layer = layerProbeFailure(processProbeFailure);
421
422 return Effect.gen(function* () {
423 const scanner = yield* PortScanner.PortDiscovery;
424 yield* TestClock.adjust(Duration.seconds(15));
425 expect(pollSpans).toBe(0);
426
427 yield* scanner.retain;
428 expect(pollSpans).toBe(1);
429 }).pipe(Effect.scoped, Effect.provide(layer), Effect.withTracer(tracer));
430});
431
432effectIt.effect("uses the current configured fragment when readiness comes from cache", () => {
433 const requests: string[] = [];
434 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
435 requests.push(String(input));
436 return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } }));
437 }) as typeof globalThis.fetch;
438 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
439 const oldUrl = `http://localhost:${LSOF_TEST_PORT}/docs#old`;
440 const newUrl = `http://localhost:${LSOF_TEST_PORT}/docs#new`;
441
442 return Effect.gen(function* () {
443 const scanner = yield* PortScanner.PortDiscovery;
444 expect((yield* scanner.scan([oldUrl]))[0]?.url).toBe(oldUrl);
445 const requestCount = requests.length;
446 expect((yield* scanner.scan([newUrl]))[0]?.url).toBe(newUrl);
447 expect(requests).toHaveLength(requestCount);
448 }).pipe(Effect.provide(layer));
449});
450
451effectIt.effect("shares a configured root probe with discovered-root classification", () => {
452 const requests: string[] = [];
453 const rootUrl = `http://localhost:${LSOF_TEST_PORT}/`;
454 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
455 requests.push(String(input));
456 return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } }));
457 }) as typeof globalThis.fetch;
458 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
459
460 return Effect.gen(function* () {
461 const scanner = yield* ownedScanner();
462 expect(yield* scanner.scan([rootUrl])).toHaveLength(1);
463 expect(requests).toEqual([rootUrl]);
464
465 yield* TestClock.adjust(Duration.seconds(15));
466 expect(yield* scanner.scan([rootUrl])).toHaveLength(1);
467 expect(requests).toEqual([rootUrl, rootUrl]);
468 }).pipe(Effect.provide(layer));
469});
470
471effectIt.effect("starts fresh cache entries after the probing batch completes", () =>
472 Effect.gen(function* () {
473 const baseClock = yield* Clock.Clock;
474 const times = [0, 20_000, 20_000, 20_000];
475 let timeIndex = 0;
476 const currentTimeMillis = () => times[Math.min(timeIndex++, times.length - 1)]!;
477 const clock: Clock.Clock = {
478 ...baseClock,
479 currentTimeMillisUnsafe: currentTimeMillis,
480 currentTimeMillis: Effect.sync(currentTimeMillis),
481 };
482 const requests: string[] = [];
483 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
484 requests.push(String(input));
485 return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } }));
486 }) as typeof globalThis.fetch;
487 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
488
489 yield* Effect.gen(function* () {
490 const scanner = yield* ownedScanner();
491 expect(yield* scanner.scan()).toHaveLength(1);
492 expect(yield* scanner.scan()).toHaveLength(1);
493 expect(requests).toHaveLength(1);
494 }).pipe(Effect.provide(layer), Effect.provideService(Clock.Clock, clock));
495 }),
496);
497
498effectIt.effect("caches a failed web probe until its bounded cache entry expires", () => {
499 let responds = false;
500 const requests: string[] = [];
501 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
502 requests.push(String(input));
503 return responds
504 ? Promise.resolve(new Response("hello", { headers: { "content-type": "text/html" } }))
505 : Promise.reject(new TypeError("not HTTP"));
506 }) as typeof globalThis.fetch;
507 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
508
509 return Effect.gen(function* () {
510 const scanner = yield* ownedScanner();
511 expect(yield* scanner.scan()).toHaveLength(0);
512 expect(yield* scanner.scan()).toHaveLength(0);
513 expect(requests).toHaveLength(2);
514
515 responds = true;
516 yield* TestClock.adjust(Duration.seconds(15));
517 expect(yield* scanner.scan()).toHaveLength(1);
518 expect(requests).toHaveLength(3);
519 }).pipe(Effect.provide(layer));
520});
521
522effectIt.effect("falls back to HTTPS and does not follow redirects while probing", () => {
523 const redirects: Array<string | undefined> = [];
524 const fetchFn = (async (
525 input: Parameters<typeof globalThis.fetch>[0],
526 init?: Parameters<typeof globalThis.fetch>[1],
527 ) => {
528 redirects.push(init?.redirect);
529 if (String(input).startsWith("http:")) throw new TypeError("TLS listener");
530 return new Response(null, { status: 302, headers: { location: "https://example.com" } });
531 }) as typeof globalThis.fetch;
532 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
533
534 return Effect.gen(function* () {
535 const scanner = yield* ownedScanner();
536 const servers = yield* scanner.scan();
537 expect(servers).toHaveLength(1);
538 expect(servers[0]?.url).toBe(`https://localhost:${LSOF_TEST_PORT}`);
539 expect(redirects).toEqual(["manual", "manual"]);
540 }).pipe(Effect.provide(layer));
541});
542
543effectIt.effect("probes a port only while T3 terminals own every listener on it", () => {
544 const owned = `p1234\ncnode\nn[::1]:${LSOF_TEST_PORT}\n`;
545 const unowned = `p5678\ncrpc\nn127.0.0.1:${LSOF_TEST_PORT}\n`;
546 let stdout = unowned;
547 const requests: string[] = [];
548 const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
549 requests.push(String(input));
550 return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } }));
551 }) as typeof globalThis.fetch;
552 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn, stdout: () => stdout });
553
554 return Effect.gen(function* () {
555 const scanner = yield* ownedScanner();
556 expect(yield* scanner.scan()).toHaveLength(0);
557 stdout = `${owned}${unowned}`;
558 expect(yield* scanner.scan()).toHaveLength(0);
559 expect(requests).toEqual([]);
560
561 stdout = owned;
562 expect(yield* scanner.scan()).toHaveLength(1);
563 expect(requests).toEqual([`http://localhost:${LSOF_TEST_PORT}/`]);
564
565 yield* scanner.unregisterTerminal({
566 threadId: "scanner-thread",
567 terminalId: "scanner-terminal",
568 });
569 expect(yield* scanner.scan()).toHaveLength(0);
570 expect(requests).toHaveLength(1);
571 }).pipe(Effect.provide(layer));
572});
573
574effectIt.effect(
575 "excludes HTTP errors, non-navigation responses, and successful non-documents",
576 () => {
577 let pid = 1;
578 let makeResponse = () =>
579 new Response("not found", { status: 404, headers: { "content-type": "text/html" } });
580 const fetchFn = ((_input: Parameters<typeof globalThis.fetch>[0]) =>
581 Promise.resolve(makeResponse())) as typeof globalThis.fetch;
582 const layer = layerLsofScanner({ pid: () => pid, fetch: fetchFn });
583
584 return Effect.gen(function* () {
585 const scanner = yield* ownedScanner([1, 2, 3, 4, 5, 6, 7]);
586 expect(yield* scanner.scan()).toHaveLength(0);
587
588 pid += 1;
589 makeResponse = () =>
590 new Response("{}", { status: 200, headers: { "content-type": "application/json" } });
591 expect(yield* scanner.scan()).toHaveLength(0);
592
593 pid += 1;
594 makeResponse = () =>
595 new Response("ready", { status: 200, headers: { "content-type": "text/plain" } });
596 expect(yield* scanner.scan()).toHaveLength(0);
597
598 pid += 1;
599 makeResponse = () => new Response(null, { status: 304, headers: { location: "/cached" } });
600 expect(yield* scanner.scan()).toHaveLength(0);
601
602 pid += 1;
603 makeResponse = () =>
604 new Response(null, { status: 204, headers: { "content-type": "text/html" } });
605 expect(yield* scanner.scan()).toHaveLength(0);
606
607 pid += 1;
608 makeResponse = () => new Response(null, { status: 302 });
609 expect(yield* scanner.scan()).toHaveLength(0);
610
611 pid += 1;
612 makeResponse = () =>
613 new Response("<html />", {
614 status: 200,
615 headers: { "content-type": "application/xhtml+xml; charset=utf-8" },
616 });
617 expect(yield* scanner.scan()).toHaveLength(1);
618 }).pipe(Effect.provide(layer));
619 },
620);
621
622effectIt.effect("aborts HTTP and HTTPS probes when they time out", () => {
623 const aborted: string[] = [];
624 const fetchFn = ((
625 input: Parameters<typeof globalThis.fetch>[0],
626 init?: Parameters<typeof globalThis.fetch>[1],
627 ) =>
628 new Promise<Response>((_resolve, reject) => {
629 const signal = init?.signal;
630 const onAbort = () => {
631 aborted.push(String(input));
632 reject(new DOMException("Aborted", "AbortError"));
633 };
634 if (signal?.aborted) {
635 onAbort();
636 } else {
637 signal?.addEventListener("abort", onAbort, { once: true });
638 }
639 })) as typeof globalThis.fetch;
640 const layer = layerLsofScanner({ pid: () => 1234, fetch: fetchFn });
641
642 return Effect.gen(function* () {
643 const scanner = yield* ownedScanner();
644 const scanFiber = yield* Effect.forkChild(scanner.scan());
645 yield* TestClock.adjust(Duration.seconds(2));
646 expect(yield* Fiber.join(scanFiber)).toHaveLength(0);
647 expect(aborted).toEqual([
648 `http://localhost:${LSOF_TEST_PORT}/`,
649 `https://localhost:${LSOF_TEST_PORT}/`,
650 ]);
651 }).pipe(Effect.provide(layer));
652});
653
654effectIt.effect("does not swallow process probe defects", () =>
655 Effect.gen(function* () {
656 const defect = new Error("unexpected process probe defect");
657 const layer = layerProbeFailure(() => Effect.die(defect));
658
659 const exit = yield* Effect.flatMap(PortScanner.PortDiscovery, (scanner) => scanner.scan()).pipe(
660 Effect.provide(layer),
661 Effect.exit,
662 );
663
664 expect(Exit.isFailure(exit)).toBe(true);
665 if (Exit.isFailure(exit)) {
666 expect(Cause.hasDies(exit.cause)).toBe(true);
667 expect(Cause.squash(exit.cause)).toBe(defect);
668 }
669 }),
670);
671
672// /proc/net/tcp: 127.0.0.1:8765 and 0.0.0.0:22 listening, an established
673// connection, and a non-loopback listener; /proc/net/tcp6 adds [::]:3001.
674const PROC_NET_TCP = ` sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode
675 0: 0100007F:223D 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 5001 1 0000000000000000 100 0 0 10 0
676 1: 00000000:0016 00000000:0000 0A 00000000:00000000 00:00000000 00000000 0 0 5002 1 0000000000000000 100 0 0 10 0
677 2: 0100007F:223D 0100007F:C350 01 00000000:00000000 00:00000000 00000000 1000 0 5003 1 0000000000000000 20 4 30 10 -1
678 3: 0A00000F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 5005 1 0000000000000000 100 0 0 10 0
679`;
680const PROC_NET_TCP6 = ` sl local_address remote_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode
681 0: 00000000000000000000000000000000:0BB9 00000000000000000000000000000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 5004 1 0000000000000000 100 0 0 10 0
682`;
683
684effectIt.effect(
685 "reads Linux listeners from /proc when lsof is missing, and stops spawning it",
686 () => {
687 let lsofSpawns = 0;
688 const fdWalks: string[] = [];
689 const procFiles: Record<string, string> = {
690 "/proc/net/tcp": PROC_NET_TCP,
691 "/proc/net/tcp6": PROC_NET_TCP6,
692 "/proc/4242/comm": "python3\n",
693 };
694 const procFds: Record<string, Record<string, string>> = {
695 "4242": { "0": "/dev/null", "3": "socket:[5001]" },
696 "77": { "5": "socket:[9999]" },
697 };
698 const missing = FileSystem.makeNoop({});
699 const fileSystem = FileSystem.layerNoop({
700 readFileString: (path) => {
701 const content = procFiles[path];
702 return content === undefined ? missing.readFileString(path) : Effect.succeed(content);
703 },
704 readDirectory: (path) => {
705 if (path === "/proc") return Effect.succeed(["self", "77", "4242"]);
706 const pid = /^\/proc\/(\d+)\/fd$/.exec(path)?.[1];
707 const fds = pid === undefined ? undefined : procFds[pid];
708 if (pid === undefined || fds === undefined) return missing.readDirectory(path);
709 fdWalks.push(pid);
710 return Effect.succeed(Object.keys(fds));
711 },
712 readLink: (path) => {
713 const [, pid, fd] = /^\/proc\/(\d+)\/fd\/(\d+)$/.exec(path) ?? [];
714 const target = pid && fd ? procFds[pid]?.[fd] : undefined;
715 return target === undefined ? missing.readLink(path) : Effect.succeed(target);
716 },
717 });
718 const fetchFn = (() =>
719 Promise.resolve(
720 new Response("hello", { headers: { "content-type": "text/html" } }),
721 )) as typeof globalThis.fetch;
722 const layer = layerProbeFailure(
723 (input) => {
724 lsofSpawns += 1;
725 return processProbeFailure(input);
726 },
727 fetchFn,
728 fileSystem,
729 );
730
731 return Effect.gen(function* () {
732 const scanner = yield* ownedScanner([4242]);
733 const first = yield* scanner.scan();
734 // Ports 22 and 3001 have no known owner, so they never receive probes.
735 expect(first.map(({ port, pid, processName }) => ({ port, pid, processName }))).toEqual([
736 { port: 8765, pid: 4242, processName: "python3" },
737 ]);
738 yield* scanner.scan();
739 expect(lsofSpawns).toBe(1);
740 // Ports 22 and 3001 have no readable owner, so the second scan walks
741 // again; it stops at 77, which holds no listener, once nothing is left.
742 expect(fdWalks).toEqual(["77", "4242", "77", "4242"]);
743 yield* scanner.scan();
744 yield* scanner.scan();
745 // Unresolved owners are retried a bounded number of times.
746 expect(fdWalks).toEqual(["77", "4242", "77", "4242", "77", "4242"]);
747 }).pipe(Effect.provide(layer));
748 },
749);
750
751effectIt.effect("keeps spawning lsof after a spawn failure that is not a missing command", () => {
752 let lsofSpawns = 0;
753 const layer = layerProbeFailure((input) => {
754 lsofSpawns += 1;
755 return Effect.fail(
756 new ProcessRunner.ProcessSpawnError({
757 command: input.command,
758 argumentCount: input.args.length,
759 cwd: input.cwd,
760 cause: PlatformError.systemError({
761 _tag: "Unknown",
762 module: "ChildProcess",
763 method: "spawn",
764 description: "EAGAIN",
765 }),
766 }),
767 );
768 });
769 return Effect.gen(function* () {
770 const scanner = yield* PortScanner.PortDiscovery;
771 yield* scanner.scan();
772 yield* scanner.scan();
773 expect(lsofSpawns).toBe(2);
774 }).pipe(Effect.provide(layer));
775});
776
777effectIt.effect("does not swallow process probe interruption", () =>
778 Effect.gen(function* () {
779 const layer = layerProbeFailure(() => Effect.interrupt);
780
781 const exit = yield* Effect.flatMap(PortScanner.PortDiscovery, (scanner) => scanner.scan()).pipe(
782 Effect.provide(layer),
783 Effect.exit,
784 );
785
786 expect(Exit.isFailure(exit)).toBe(true);
787 if (Exit.isFailure(exit)) {
788 expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true);
789 }
790 }),
791);