1// @effect-diagnostics nodeBuiltinImport:off - Exercise the actual standalone Node preload and TCP transport.
2import * as NodeChildProcess from "node:child_process";
3import * as NodeEvents from "node:events";
4import * as NodeFSP from "node:fs/promises";
5import * as NodeNet from "node:net";
6import * as NodeOS from "node:os";
7import * as NodePath from "node:path";
8import * as NodeURL from "node:url";
9import { afterEach, describe, expect, it } from "vite-plus/test";
10
11import { decodeEventStream, Root, type StreamEvent } from "../../clo-lib/progress.ts";
12import { progressPreloadSource } from "./progressPreload.ts";
13
14interface Packet {
15 type: "hello" | "progress";
16 token?: string;
17 pid?: number;
18 producerId?: string;
19 streamId?: string;
20 event?: StreamEvent;
21}
22
23const cleanups: Array<() => Promise<unknown>> = [];
24afterEach(async () => {
25 for (const cleanup of cleanups.splice(0).toReversed()) await cleanup();
26});
27
28async function fixture() {
29 const directory = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3 progress preload "));
30 cleanups.push(() => NodeFSP.rm(directory, { recursive: true, force: true }));
31 const preload = NodePath.join(directory, "preload.cjs");
32 await NodeFSP.writeFile(preload, progressPreloadSource);
33 const packets: Packet[] = [];
34 const sockets = new Set<NodeNet.Socket>();
35 const changes = new Set<() => void>();
36 const server = NodeNet.createServer((socket) => {
37 sockets.add(socket);
38 socket.on("close", () => sockets.delete(socket));
39 socket.on("error", () => {});
40 let pending = "";
41 socket.on("data", (chunk) => {
42 pending += chunk.toString();
43 let index: number;
44 while ((index = pending.indexOf("\n")) !== -1) {
45 packets.push(JSON.parse(pending.slice(0, index)) as Packet);
46 pending = pending.slice(index + 1);
47 }
48 for (const changed of changes) changed();
49 });
50 });
51 server.listen(0, "127.0.0.1");
52 await NodeEvents.EventEmitter.once(server, "listening");
53 const address = server.address();
54 if (!address || typeof address === "string") throw new Error("Missing listener port");
55 cleanups.push(async () => {
56 for (const socket of sockets) socket.destroy();
57 await new Promise<void>((resolve) => server.close(() => resolve()));
58 });
59 const environment = {
60 ...process.env,
61 NODE_OPTIONS: `--require=${JSON.stringify(preload)}`,
62 T3_SCRIPT_PROGRESS_ENDPOINT: `127.0.0.1:${address.port}`,
63 T3_SCRIPT_PROGRESS_TOKEN: "test-progress-token",
64 };
65 const waitFor = (predicate: (rows: readonly Packet[]) => boolean) => {
66 if (predicate(packets)) return Promise.resolve();
67 return new Promise<void>((resolve) => {
68 const changed = () => {
69 if (!predicate(packets)) return;
70 changes.delete(changed);
71 resolve();
72 };
73 changes.add(changed);
74 });
75 };
76 let scriptNumber = 0;
77 function run(
78 source: string,
79 options: { endpoint?: boolean; shebang?: boolean; preload?: boolean } = {},
80 ) {
81 const script = NodePath.join(directory, `script-${++scriptNumber}.mjs`);
82 const childReady = Promise.withResolvers<void>();
83 const children: NodeChildProcess.ChildProcess[] = [];
84 const completion = (async () => {
85 await NodeFSP.writeFile(
86 script,
87 `${options.shebang ? `#!${process.execPath}\n` : ""}${source}`,
88 );
89 if (options.shebang) await NodeFSP.chmod(script, 0o700);
90 const child = NodeChildProcess.spawn(
91 options.shebang ? script : process.execPath,
92 options.shebang ? [] : [script],
93 {
94 cwd: NodeOS.tmpdir(),
95 env: {
96 ...environment,
97 ...(options.endpoint === false ? { T3_SCRIPT_PROGRESS_ENDPOINT: "" } : {}),
98 ...(options.preload === false ? { NODE_OPTIONS: "" } : {}),
99 },
100 stdio: ["ignore", "pipe", "pipe", "ipc"],
101 },
102 );
103 children.push(child);
104 child.on("message", () => childReady.resolve());
105 let stdout = "";
106 let stderr = "";
107 child.stdout?.on("data", (chunk) => {
108 stdout += chunk;
109 });
110 child.stderr?.on("data", (chunk) => {
111 stderr += chunk;
112 });
113 const [code] = await NodeEvents.EventEmitter.once(child, "close");
114 return { code, stdout, stderr };
115 })();
116 cleanups.push(async () => {
117 for (const child of children)
118 if (child.exitCode === null && child.signalCode === null) child.kill();
119 await completion;
120 });
121 return {
122 completion,
123 ready: childReady.promise,
124 send: () => {
125 const child = children[0];
126 if (!child) throw new Error("Child has not started");
127 child.send("continue");
128 },
129 };
130 }
131 return { directory, environment, packets, sockets, waitFor, run };
132}
133
134const localCopies: string[] = JSON.parse(process.env.T3_PROGRESS_TEST_MODULES ?? "[]");
135const copies = [
136 NodeURL.fileURLToPath(new URL("../../clo-lib/progress.ts", import.meta.url)),
137 ...localCopies,
138];
139const hasNode = (packets: readonly Packet[], label: string) =>
140 packets.some((packet) =>
141 packet.event?.some(
142 (part) =>
143 typeof part === "object" && part !== null && !Array.isArray(part) && part.t === label,
144 ),
145 );
146
147async function decodeSnapshot(events: StreamEvent[]) {
148 const root = new Root();
149 let controller: ReadableStreamDefaultController<StreamEvent>;
150 const input = new ReadableStream<StreamEvent>({
151 start(value) {
152 controller = value;
153 },
154 });
155 const decoding = decodeEventStream<void, { checkpoint: [] }>(input, root);
156 const ready = new Promise<void>((resolve) => decoding.on("checkpoint", resolve));
157 for (const event of events) controller!.enqueue(event);
158 controller!.enqueue([0, ["checkpoint"]]);
159 await ready;
160 const result = root.active.map((node) => ({
161 text: node.text,
162 value: node.value,
163 children: node.children.map((child) => ({ text: child.text, value: child.value })),
164 }));
165 decoding.cancel();
166 return result;
167}
168
169describe.each(copies)("progress preload using %s", (copy) => {
170 it("captures global and custom roots, nested updates, and preserves normal output", async () => {
171 const f = await fixture();
172 const child = f.run(`
173 const p = await import(${JSON.stringify(NodeURL.pathToFileURL(copy).href)});
174 p.start("global task");
175 const root = new p.Root();
176 const parent = root.start("custom task", { total: 10, value: 4 });
177 parent.start("nested task", { total: 3, value: 1 });
178 console.log("script stdout");
179 console.error("script stderr");
180 process.send("ready");
181 await new Promise(resolve => process.once("message", resolve));
182 parent.value = 0;
183 parent.text = "";
184 process.send("updated");
185 await new Promise(resolve => process.once("message", resolve));
186 process.disconnect();
187 `);
188 await child.ready;
189 await f.waitFor((rows) => hasNode(rows, "global task") && hasNode(rows, "nested task"));
190 expect(f.packets.find((packet) => packet.type === "hello")).toMatchObject({
191 token: "test-progress-token",
192 producerId: expect.any(String),
193 });
194 child.send();
195 await f.waitFor((rows) =>
196 rows.some((row) =>
197 row.event?.some(
198 (part) =>
199 typeof part === "object" &&
200 part !== null &&
201 !Array.isArray(part) &&
202 part.t === "" &&
203 part.v === 0,
204 ),
205 ),
206 );
207 const stream = f.packets.find((packet) => hasNode([packet], "custom task"))?.streamId;
208 expect(stream).toBeDefined();
209 const events = f.packets.flatMap((packet) =>
210 packet.streamId === stream && packet.event ? [packet.event] : [],
211 );
212 expect(await decodeSnapshot(events)).toEqual([
213 { text: "", value: 0, children: [{ text: "nested task", value: 1 }] },
214 ]);
215 child.send();
216 const baseline = f.run(
217 `const p = await import(${JSON.stringify(NodeURL.pathToFileURL(copy).href)}); p.start("global task"); console.log("script stdout"); console.error("script stderr"); process.disconnect();`,
218 { preload: false },
219 );
220 expect(await child.completion).toEqual(await baseline.completion);
221 });
222
223 it("fails open without a configured endpoint", async () => {
224 const f = await fixture();
225 const child = f.run(
226 `const p = await import(${JSON.stringify(NodeURL.pathToFileURL(copy).href)}); p.start("unobserved"); console.log("unchanged"); process.disconnect();`,
227 { endpoint: false },
228 );
229 const baseline = f.run(
230 `const p = await import(${JSON.stringify(NodeURL.pathToFileURL(copy).href)}); p.start("unobserved"); console.log("unchanged"); process.disconnect();`,
231 { preload: false },
232 );
233 expect(await child.completion).toEqual(await baseline.completion);
234 expect(f.packets).toHaveLength(0);
235 });
236});
237
238it("captures inherited preload in nested absolute Node and executable shebang scripts", async () => {
239 const f = await fixture();
240 const nested = NodePath.join(f.directory, "nested.mjs");
241 await NodeFSP.writeFile(
242 nested,
243 `import { start } from ${JSON.stringify(NodeURL.pathToFileURL(copies[0]!).href)}; start("nested process"); console.log("nested stdout");`,
244 );
245 const child = f.run(
246 `
247 import { spawn } from "node:child_process";
248 const child = spawn(process.execPath, [${JSON.stringify(nested)}], { stdio: ["ignore", "inherit", "inherit"] });
249 child.on("exit", code => { process.exitCode = code; process.disconnect(); });
250 `,
251 { shebang: true },
252 );
253 expect(await child.completion).toEqual({ code: 0, stdout: "nested stdout\n", stderr: "" });
254 expect(hasNode(f.packets, "nested process")).toBe(true);
255 expect(f.packets.filter((packet) => packet.type === "hello")).toHaveLength(1);
256});
257
258it("leaves unrelated Node modules and exit status alone", async () => {
259 const f = await fixture();
260 const child = f.run(
261 `import { basename } from "node:path"; console.log(basename("/some/file")); process.disconnect(); process.exit(7);`,
262 );
263 expect(await child.completion).toEqual({ code: 7, stdout: "file\n", stderr: "" });
264 expect(f.packets).toHaveLength(0);
265});
266
267it("resnapshots active roots after the collector connection is replaced", async () => {
268 const f = await fixture();
269 const child = f.run(`
270 import { Root } from ${JSON.stringify(NodeURL.pathToFileURL(copies[0]!).href)};
271 const root = new Root();
272 root.start("reconnect parent", { total: 8, value: 3 }).start("reconnect child");
273 process.send("ready");
274 await new Promise(resolve => process.once("message", resolve));
275 process.disconnect();
276 `);
277 await child.ready;
278 await f.waitFor((rows) => hasNode(rows, "reconnect child"));
279 const firstCount = f.packets.length;
280 for (const socket of f.sockets) socket.destroy();
281 await f.waitFor(
282 (rows) =>
283 rows.filter((row) => row.type === "hello").length === 2 &&
284 hasNode(rows.slice(firstCount), "reconnect child"),
285 );
286 expect(
287 new Set(f.packets.filter((row) => row.type === "hello").map((row) => row.producerId)).size,
288 ).toBe(1);
289 const snapshot = f.packets.slice(firstCount).find((row) => row.event)?.event;
290 expect(snapshot?.[0]).toBe(-1);
291 expect(await decodeSnapshot(snapshot ? [snapshot] : [])).toEqual([
292 { text: "reconnect parent", value: 3, children: [{ text: "reconnect child", value: 0 }] },
293 ]);
294 child.send();
295 expect((await child.completion).code).toBe(0);
296});
297
298it("observes separate loaded copies without sharing root identities", async () => {
299 const f = await fixture();
300 const url = NodeURL.pathToFileURL(copies[0]!).href;
301 const child = f.run(`
302 const first = await import(${JSON.stringify(url)});
303 const second = await import(${JSON.stringify(`${url}?second-copy`)});
304 first.start("first copy");
305 second.start("second copy");
306 process.send("ready");
307 await new Promise(resolve => process.once("message", resolve));
308 process.disconnect();
309 `);
310 await child.ready;
311 await f.waitFor((rows) => hasNode(rows, "first copy") && hasNode(rows, "second copy"));
312 const ids = f.packets
313 .filter((row) => hasNode([row], "first copy") || hasNode([row], "second copy"))
314 .map((row) => row.streamId);
315 expect(new Set(ids).size).toBe(2);
316 expect(f.packets.filter((row) => row.type === "hello")).toHaveLength(1);
317 child.send();
318 expect((await child.completion).code).toBe(0);
319});
320
321it("fails open when the Node runtime cannot install synchronous module hooks", async () => {
322 const f = await fixture();
323 const shim = NodePath.join(f.directory, "old-runtime.cjs");
324 await NodeFSP.writeFile(
325 shim,
326 `const modules = require("node:module"); modules.registerHooks = undefined; modules.syncBuiltinESMExports();`,
327 );
328 f.environment.NODE_OPTIONS = `--require=${JSON.stringify(shim)} ${f.environment.NODE_OPTIONS}`;
329 const child = f.run(
330 `import { start } from ${JSON.stringify(NodeURL.pathToFileURL(copies[0]!).href)}; start("not captured"); console.log("still works"); process.disconnect();`,
331 );
332 expect(await child.completion).toEqual({ code: 0, stdout: "still works\n", stderr: "" });
333 expect(f.packets).toHaveLength(0);
334});
335
336it("preserves genuine parent FD forwarding while capturing an independent child root", async () => {
337 const f = await fixture();
338 const url = NodeURL.pathToFileURL(copies[0]!).href;
339 const nested = NodePath.join(f.directory, "forwarded.mjs");
340 await NodeFSP.writeFile(
341 nested,
342 `
343 import * as p from ${JSON.stringify(url)};
344 p.start("forwarded global");
345 new p.Root().start("independent child");
346 process.send("ready");
347 await new Promise(resolve => process.once("message", resolve));
348 process.disconnect();
349 `,
350 );
351 const child = f.run(`
352 import { spawn } from "node:child_process";
353 import { Readable } from "node:stream";
354 import * as p from ${JSON.stringify(url)};
355 const parent = new p.Root().start("parent owns progress");
356 const child = spawn(process.execPath, [${JSON.stringify(nested)}], {
357 env: { ...process.env, ...p.ipcChildEnvironment(3) },
358 stdio: ["ignore", "inherit", "inherit", "pipe", "ipc"],
359 });
360 const decoding = p.decodeAttachToRef(Readable.toWeb(child.stdio[3]), parent);
361 child.on("message", () => process.send("ready"));
362 await new Promise(resolve => process.once("message", resolve));
363 child.send("finish");
364 await new Promise(resolve => child.once("close", resolve));
365 await decoding;
366 process.disconnect();
367 `);
368 await child.ready;
369 await f.waitFor(
370 (rows) => hasNode(rows, "forwarded global") && hasNode(rows, "independent child"),
371 );
372 const forwarded = f.packets.filter((row) => hasNode([row], "forwarded global"));
373 expect(forwarded).toHaveLength(1);
374 const parentStream = f.packets.find((row) => hasNode([row], "parent owns progress"))?.streamId;
375 expect(forwarded[0]?.streamId).toBe(parentStream);
376 expect(f.packets.filter((row) => row.type === "hello")).toHaveLength(2);
377 child.send();
378 expect((await child.completion).code).toBe(0);
379});
380
381it("resolves package imports from the script location independently of the current directory", async () => {
382 const f = await fixture();
383 const packageDirectory = NodePath.join(f.directory, "node_modules", "@clo");
384 await NodeFSP.mkdir(packageDirectory, { recursive: true });
385 await NodeFSP.symlink(
386 NodePath.dirname(copies[0]!),
387 NodePath.join(packageDirectory, "lib"),
388 "dir",
389 );
390 const child = f.run(
391 `import { start } from "@clo/lib/progress"; start("package import"); process.send("ready"); await new Promise(resolve => process.once("message", resolve)); process.disconnect();`,
392 );
393 await child.ready;
394 await f.waitFor((rows) => hasNode(rows, "package import"));
395 child.send();
396 expect((await child.completion).code).toBe(0);
397});
398
399describe.each(copies)("completed root using %s", (copy) => {
400 it("sends exactly one terminal event and closes the observed root", async () => {
401 const f = await fixture();
402 const child = f.run(`
403 const { Root } = await import(${JSON.stringify(NodeURL.pathToFileURL(copy).href)});
404 const root = new Root();
405 root.start("finite custom task");
406 process.send("ready");
407 await new Promise(resolve => process.once("message", resolve));
408 root.end("finished");
409 process.send("ended");
410 await new Promise(resolve => process.once("message", resolve));
411 process.disconnect();
412 `);
413 await child.ready;
414 await f.waitFor((rows) => hasNode(rows, "finite custom task"));
415 child.send();
416 await f.waitFor((rows) =>
417 rows.some((row) => row.event?.some((part) => Array.isArray(part) && part[0] === "end")),
418 );
419 const terminals = f.packets.flatMap(
420 (row) => row.event?.filter((part) => Array.isArray(part) && part[0] === "end") ?? [],
421 );
422 expect(terminals).toEqual([["end", "finished"]]);
423 child.send();
424 expect((await child.completion).code).toBe(0);
425 });
426});
427
428describe.each(copies)("error semantics using %s", (copy) => {
429 it("preserves the original unhandled root error and nonzero exit status", async () => {
430 const f = await fixture();
431 const source = `
432 const { Root } = await import(${JSON.stringify(NodeURL.pathToFileURL(copy).href)});
433 const root = new Root();
434 root.start("task");
435 root.error(new Error("deliberate failure"));
436 console.log("SURVIVED");
437 `;
438 const child = f.run(source);
439 const baseline = f.run(source, { preload: false });
440 const original = await baseline.completion;
441 const observed = await child.completion;
442 expect(original.code).toBe(1);
443 expect(observed.code).toBe(original.code);
444 expect(observed.stdout).toBe(original.stdout);
445 expect(observed.stdout).not.toContain("SURVIVED");
446 expect(observed.stderr).toContain("deliberate failure");
447 });
448
449 it("preserves existing handled root errors", async () => {
450 const f = await fixture();
451 const source = `
452 const { Root } = await import(${JSON.stringify(NodeURL.pathToFileURL(copy).href)});
453 const root = new Root();
454 root.on("error", () => console.log("handled"));
455 root.start("task");
456 root.error(new Error("expected failure"));
457 console.log("continued");
458 process.disconnect();
459 `;
460 const child = f.run(source);
461 const baseline = f.run(source, { preload: false });
462 expect(await child.completion).toEqual(await baseline.completion);
463 expect((await child.completion).stdout).toBe("handled\ncontinued\n");
464 });
465});
466
467it("captures a new root after reconnecting and then losing an idle collector connection", async () => {
468 const f = await fixture();
469 const child = f.run(`
470 import { Root } from ${JSON.stringify(NodeURL.pathToFileURL(copies[0]!).href)};
471 const first = new Root();
472 first.start("first active root");
473 process.send("ready");
474 await new Promise(resolve => process.once("message", resolve));
475 first.end();
476 await new Promise(resolve => process.once("message", resolve));
477 new Root().start("later root");
478 process.send("ready");
479 await new Promise(resolve => process.once("message", resolve));
480 process.disconnect();
481 `);
482 await child.ready;
483 await f.waitFor((rows) => hasNode(rows, "first active root"));
484 for (const socket of f.sockets) socket.destroy();
485 await f.waitFor((rows) => rows.filter((row) => row.type === "hello").length === 2);
486 child.send();
487 await f.waitFor((rows) =>
488 rows.some((row) => row.event?.some((part) => Array.isArray(part) && part[0] === "end")),
489 );
490 await Promise.all(
491 [...f.sockets].map(async (socket) => {
492 const closed = NodeEvents.EventEmitter.once(socket, "close");
493 socket.destroy();
494 await closed;
495 }),
496 );
497 child.send();
498 await f.waitFor(
499 (rows) =>
500 rows.filter((row) => row.type === "hello").length === 3 && hasNode(rows, "later root"),
501 );
502 child.send();
503 expect((await child.completion).code).toBe(0);
504});
505
506it("installs hooks before an existing require preload imports progress", async () => {
507 const f = await fixture();
508 const userPreload = NodePath.join(f.directory, "existing user preload.cjs");
509 await NodeFSP.writeFile(
510 userPreload,
511 `const progress = require(${JSON.stringify(copies[0]!)}); progress.start("user preload task");`,
512 );
513 f.environment.NODE_OPTIONS += ` --require=${JSON.stringify(userPreload)}`;
514 const child = f.run(`
515 import { start } from ${JSON.stringify(NodeURL.pathToFileURL(copies[0]!).href)};
516 start("main task after preload");
517 process.send("ready");
518 await new Promise(resolve => process.once("message", resolve));
519 process.disconnect();
520 `);
521 await child.ready;
522 await f.waitFor(
523 (rows) => hasNode(rows, "user preload task") && hasNode(rows, "main task after preload"),
524 );
525 child.send();
526 expect((await child.completion).code).toBe(0);
527});
528
529it("distinguishes worker-thread producers sharing one process ID", async () => {
530 const f = await fixture();
531 const url = NodeURL.pathToFileURL(copies[0]!).href;
532 const workerSource = `
533 const { parentPort } = require("node:worker_threads");
534 import(${JSON.stringify(url)}).then(({ Root }) => {
535 new Root().start("worker task");
536 parentPort.postMessage("ready");
537 parentPort.once("message", () => parentPort.close());
538 });
539 `;
540 const child = f.run(`
541 import { Worker } from "node:worker_threads";
542 import { Root } from ${JSON.stringify(url)};
543 new Root().start("main thread task");
544 const worker = new Worker(${JSON.stringify(workerSource)}, { eval: true });
545 worker.once("message", () => process.send("ready"));
546 await new Promise(resolve => process.once("message", resolve));
547 worker.postMessage("finish");
548 await new Promise(resolve => worker.once("exit", resolve));
549 process.disconnect();
550 `);
551 await child.ready;
552 await f.waitFor((rows) => hasNode(rows, "main thread task") && hasNode(rows, "worker task"));
553 const producers = f.packets.filter((packet) => packet.type === "hello");
554 expect(producers).toHaveLength(2);
555 expect(new Set(producers.map((packet) => packet.pid)).size).toBe(1);
556 expect(new Set(producers.map((packet) => packet.producerId)).size).toBe(2);
557 child.send();
558 expect((await child.completion).code).toBe(0);
559});