1// @effect-diagnostics nodeBuiltinImport:off -- A native child validates NODE_OPTIONS before application code runs.
2import * as NodeNet from "node:net";
3import * as NodeChildProcess from "node:child_process";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { expect, it } from "@effect/vitest";
6import { Root, decodeEventStream, type StreamEvent } from "@clo/lib/progress";
7import { ThreadId, type ScriptProgressEvent } from "@t3tools/contracts";
8import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
9import * as Effect from "effect/Effect";
10import * as FileSystem from "effect/FileSystem";
11import * as Layer from "effect/Layer";
12import * as Queue from "effect/Queue";
13import * as Stream from "effect/Stream";
14
15import * as ServerConfig from "../config.ts";
16import * as Capture from "./ScriptProgress.ts";
17
18const threadId = ThreadId.make("progress-thread");
19const otherThreadId = ThreadId.make("other-progress-thread");
20const testLayer = Capture.layer.pipe(
21 Layer.provide(ServerConfig.layerTest("/tmp", { prefix: "t3 script progress " })),
22 Layer.provideMerge(NodeServices.layer),
23);
24
25const connect = (endpoint: string) =>
26 Effect.acquireRelease(
27 Effect.callback<NodeNet.Socket>((resume) => {
28 const socket = NodeNet.connect({ host: "127.0.0.1", port: Number(endpoint.split(":")[1]) });
29 socket.once("connect", () => resume(Effect.succeed(socket)));
30 socket.once("error", (error) => resume(Effect.die(error)));
31 return Effect.sync(() => socket.destroy());
32 }),
33 (socket) => Effect.sync(() => socket.destroy()),
34 );
35
36const closed = (socket: NodeNet.Socket) =>
37 Effect.callback<void>((resume) => {
38 if (socket.closed) return resume(Effect.void);
39 socket.once("close", () => resume(Effect.void));
40 });
41
42const write = (socket: NodeNet.Socket, value: unknown) =>
43 Effect.sync(() => {
44 socket.write(`${JSON.stringify(value)}\n`);
45 });
46
47const until = <E, R>(
48 pull: Effect.Effect<ScriptProgressEvent, E, R>,
49 predicate: (event: ScriptProgressEvent) => boolean,
50) =>
51 Effect.gen(function* () {
52 while (true) {
53 const event = yield* pull;
54 if (predicate(event)) return event;
55 }
56 });
57
58it.effect("materializes an external preload and gives each thread an isolated capture token", () =>
59 Effect.gen(function* () {
60 const progress = yield* ScriptProgress.ScriptProgress;
61 const first = yield* progress.environmentForThread(threadId);
62 const again = yield* progress.environmentForThread(threadId);
63 const other = yield* progress.environmentForThread(otherThreadId);
64 expect(first).toEqual(again);
65 expect(first.T3_SCRIPT_PROGRESS_TOKEN).not.toBe(other.T3_SCRIPT_PROGRESS_TOKEN);
66 expect(first.T3_SCRIPT_PROGRESS_ENDPOINT).toMatch(/^127\.0\.0\.1:\d+$/);
67 const fs = yield* FileSystem.FileSystem;
68 const preloadPath: unknown = JSON.parse(first.NODE_OPTIONS!.slice("--require=".length));
69 if (typeof preloadPath !== "string") throw new Error("Expected preload path");
70 const stat = yield* fs.stat(preloadPath);
71 expect(Number(stat.size)).toBeGreaterThan(100);
72 expect(preloadPath).not.toContain(".asar/");
73 expect(preloadPath).toContain("t3 script progress ");
74 expect(preloadPath).toMatch(/\.cjs$/);
75 const environment = ScriptProgress.withScriptProgressEnvironment(
76 {
77 ...process.env,
78 NODE_OPTIONS: "--max-old-space-size=128",
79 },
80 first,
81 );
82 expect(ScriptProgress.withScriptProgressEnvironment(environment, first)).toEqual(environment);
83 const result = yield* Effect.callback<{ code: number | null; stdout: string; stderr: string }>(
84 (resume) => {
85 const child = NodeChildProcess.spawn(
86 process.execPath,
87 ["-e", "process.stdout.write('preload-loaded')"],
88 { env: environment },
89 );
90 let stdout = "";
91 let stderr = "";
92 child.stdout.on("data", (chunk: Buffer) => {
93 stdout += chunk.toString();
94 });
95 child.stderr.on("data", (chunk: Buffer) => {
96 stderr += chunk.toString();
97 });
98 child.once("error", (error) => resume(Effect.die(error)));
99 child.once("close", (code) => resume(Effect.succeed({ code, stdout, stderr })));
100 return Effect.sync(() => {
101 child.kill();
102 });
103 },
104 );
105 expect(result).toEqual({ code: 0, stdout: "preload-loaded", stderr: "" });
106 }).pipe(Effect.scoped, Effect.provide(testLayer)),
107);
108
109it.effect(
110 "late subscribers get a current reset snapshot with nested nodes and subsequent zero deltas",
111 () =>
112 Effect.gen(function* () {
113 const progress = yield* ScriptProgress.ScriptProgress;
114 const env = yield* progress.environmentForThread(threadId);
115 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
116 const pull = Queue.take(queue);
117 expect(yield* pull).toEqual({ type: "snapshot", streams: [] });
118 const socket = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
119 yield* write(socket, {
120 type: "hello",
121 producerId: "fixture",
122 token: env.T3_SCRIPT_PROGRESS_TOKEN,
123 pid: 123,
124 label: "build.ts",
125 });
126 yield* write(socket, {
127 type: "progress",
128 streamId: "global",
129 event: [
130 -1,
131 { k: 1, t: "Build", v: 2, e: 10, c: [2] },
132 { k: 2, t: "Assets", v: 1, e: 3 },
133 [1],
134 ],
135 });
136 yield* until(pull, (event) => event.type === "progress" && event.summary[0]?.value === 2);
137 yield* write(socket, { type: "progress", streamId: "global", event: [0, { k: 1, v: 7 }] });
138 yield* until(pull, (event) => event.type === "progress" && event.summary[0]?.value === 7);
139
140 const lateQueue = yield* Stream.toQueue(progress.subscribe(threadId), {
141 capacity: "unbounded",
142 });
143 const late = Queue.take(lateQueue);
144 const snapshot = yield* late;
145 expect(snapshot).toMatchObject({
146 type: "snapshot",
147 streams: [
148 {
149 pid: 123,
150 label: "build.ts",
151 status: "connected",
152 summary: [{ text: "Build", value: 7, total: 10 }],
153 },
154 ],
155 });
156 const reset = yield* until(late, (event) => event.type === "progress");
157 if (reset.type !== "progress") throw new Error("Expected progress");
158 expect(reset.event[0]).toBe(-1);
159 const root = new Root();
160 let input!: ReadableStreamDefaultController<StreamEvent>;
161 const decoder = decodeEventStream(
162 new ReadableStream({
163 start(controller) {
164 input = controller;
165 },
166 }),
167 root,
168 );
169 yield* Effect.addFinalizer(() => Effect.sync(() => decoder.cancel()));
170 const decoded = new Promise<void>((resolve) =>
171 root.on("node-start", () => {
172 if (root.active[0]?.children.length === 1) resolve();
173 }),
174 );
175 input.enqueue(reset.event as StreamEvent);
176 yield* Effect.promise(() => decoded);
177 expect(root.active[0]?.value).toBe(7);
178 expect(root.active[0]?.children[0]?.text).toBe("Assets");
179 yield* write(socket, { type: "progress", streamId: "global", event: [0, { k: 1, v: 0 }] });
180 yield* until(late, (event) => event.type === "progress" && event.summary[0]?.value === 0);
181 yield* write(socket, { type: "progress", streamId: "global", event: [0, ["end", null]] });
182 yield* until(late, (event) => event.type === "remove");
183 }).pipe(Effect.scoped, Effect.provide(testLayer)),
184);
185
186it.effect("rejects unknown tokens and malformed producers without affecting another thread", () =>
187 Effect.gen(function* () {
188 const progress = yield* ScriptProgress.ScriptProgress;
189 const env = yield* progress.environmentForThread(threadId);
190 const wrong = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
191 yield* write(wrong, { type: "hello", producerId: "fixture", token: "wrong", pid: 12 });
192 yield* closed(wrong);
193 const malformed = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
194 yield* Effect.sync(() => malformed.write("not json\n"));
195 yield* closed(malformed);
196 const valid = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
197 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
198 const pull = Queue.take(queue);
199 yield* pull;
200 yield* write(valid, {
201 type: "hello",
202 producerId: "fixture",
203 token: env.T3_SCRIPT_PROGRESS_TOKEN,
204 pid: 13,
205 });
206 yield* write(valid, {
207 type: "progress",
208 streamId: "global",
209 event: [-1, { k: 1, t: "Good" }, [1]],
210 });
211 yield* until(pull, (event) => event.type === "progress" && event.summary[0]?.text === "Good");
212 const otherQueue = yield* Stream.toQueue(progress.subscribe(otherThreadId), {
213 capacity: "unbounded",
214 });
215 const other = Queue.take(otherQueue);
216 expect(yield* other).toEqual({ type: "snapshot", streams: [] });
217 }).pipe(Effect.scoped, Effect.provide(testLayer)),
218);
219
220it.effect(
221 "coalesces a burst into current progress and labels disconnected producers as unknown",
222 () =>
223 Effect.gen(function* () {
224 const progress = yield* ScriptProgress.ScriptProgress;
225 const env = yield* progress.environmentForThread(threadId);
226 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
227 const pull = Queue.take(queue);
228 yield* pull;
229 const socket = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
230 yield* write(socket, {
231 type: "hello",
232 producerId: "fixture",
233 token: env.T3_SCRIPT_PROGRESS_TOKEN,
234 pid: 14,
235 });
236 yield* write(socket, {
237 type: "progress",
238 streamId: "global",
239 event: [-1, { k: 1, t: "Burst", e: 1000 }, [1]],
240 });
241 yield* until(
242 pull,
243 (event) => event.type === "progress" && event.summary[0]?.text === "Burst",
244 );
245 yield* Effect.sync(() => {
246 socket.write(
247 Array.from({ length: 100 }, (_, i) =>
248 JSON.stringify({
249 type: "progress",
250 streamId: "global",
251 event: [0, { k: 1, v: i + 1 }],
252 }),
253 ).join("\n") + "\n",
254 );
255 });
256 yield* until(pull, (event) => event.type === "progress" && event.summary[0]?.value === 100);
257 yield* Effect.sync(() => socket.destroy());
258 const event = yield* until(
259 pull,
260 (event) => event.type === "upsert" && event.stream.status === "disconnected",
261 );
262 expect(event).toMatchObject({ type: "upsert", stream: { status: "disconnected" } });
263 const lateQueue = yield* Stream.toQueue(progress.subscribe(threadId), {
264 capacity: "unbounded",
265 });
266 const late = Queue.take(lateQueue);
267 expect(yield* late).toMatchObject({
268 type: "snapshot",
269 streams: [{ status: "disconnected", summary: [{ text: "Burst", value: 100 }] }],
270 });
271 }).pipe(Effect.scoped, Effect.provide(testLayer)),
272);
273
274it.effect("destroys oversized input without retaining a capture", () =>
275 Effect.gen(function* () {
276 const progress = yield* ScriptProgress.ScriptProgress;
277 const env = yield* progress.environmentForThread(threadId);
278 const socket = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
279 yield* write(socket, {
280 type: "hello",
281 producerId: "fixture",
282 token: env.T3_SCRIPT_PROGRESS_TOKEN,
283 pid: 15,
284 });
285 yield* Effect.sync(() => socket.write("x".repeat(1024 * 1024 + 1)));
286 yield* closed(socket);
287 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
288 const pull = Queue.take(queue);
289 expect(yield* pull).toEqual({ type: "snapshot", streams: [] });
290 }).pipe(Effect.scoped, Effect.provide(testLayer)),
291);
292
293it.effect("bounds accumulated state without treating a broken producer as completed", () =>
294 Effect.gen(function* () {
295 const progress = yield* ScriptProgress.ScriptProgress;
296 const env = yield* progress.environmentForThread(threadId);
297 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
298 const pull = Queue.take(queue);
299 yield* pull;
300 const socket = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
301 yield* write(socket, {
302 type: "hello",
303 producerId: "fixture",
304 token: env.T3_SCRIPT_PROGRESS_TOKEN,
305 pid: 16,
306 });
307 yield* write(socket, {
308 type: "progress",
309 streamId: "global",
310 event: [-1, { k: 1, t: "a".repeat(400_000) }, [1]],
311 });
312 yield* until(pull, (event) => event.type === "progress" && event.summary.length === 1);
313 yield* write(socket, {
314 type: "progress",
315 streamId: "global",
316 event: [
317 0,
318 { k: 1, c: [2, 3] },
319 { k: 2, t: "b".repeat(400_000) },
320 { k: 3, t: "c".repeat(400_000) },
321 ],
322 });
323 yield* until(pull, (event) => event.type === "remove");
324 yield* closed(socket);
325 const lateQueue = yield* Stream.toQueue(progress.subscribe(threadId), {
326 capacity: "unbounded",
327 });
328 expect(yield* Queue.take(lateQueue)).toEqual({ type: "snapshot", streams: [] });
329 }).pipe(Effect.scoped, Effect.provide(testLayer)),
330);
331
332it.effect("retains only the most recent sixteen disconnected roots for a thread", () =>
333 Effect.gen(function* () {
334 const progress = yield* ScriptProgress.ScriptProgress;
335 const env = yield* progress.environmentForThread(threadId);
336 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
337 const pull = Queue.take(queue);
338 yield* pull;
339 const producers: NodeNet.Socket[] = [];
340 for (let i = 0; i < 17; i++) {
341 const socket = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
342 yield* write(socket, {
343 type: "hello",
344 producerId: `producer-${i}`,
345 token: env.T3_SCRIPT_PROGRESS_TOKEN,
346 pid: 100 + i,
347 });
348 yield* write(socket, {
349 type: "progress",
350 streamId: "global",
351 event: [-1, { k: 1, t: "Retained" }, [1]],
352 });
353 yield* until(pull, (event) => event.type === "upsert" && event.stream.status === "connected");
354 producers.push(socket);
355 }
356 for (const socket of producers) {
357 yield* Effect.sync(() => socket.destroy());
358 yield* until(
359 pull,
360 (event) => event.type === "upsert" && event.stream.status === "disconnected",
361 );
362 }
363 const lateQueue = yield* Stream.toQueue(progress.subscribe(threadId), {
364 capacity: "unbounded",
365 });
366 const snapshot = yield* Queue.take(lateQueue);
367 if (snapshot.type !== "snapshot") throw new Error("Expected snapshot");
368 expect(snapshot.streams).toHaveLength(16);
369 expect(snapshot.streams.map((capture) => capture.pid)).toEqual(
370 Array.from({ length: 16 }, (_, i) => 101 + i),
371 );
372 }).pipe(Effect.scoped, Effect.provide(testLayer)),
373);
374
375it.effect("closes the loopback listener when its application scope ends", () =>
376 Effect.gen(function* () {
377 const endpoint = yield* Effect.gen(function* () {
378 const progress = yield* ScriptProgress.ScriptProgress;
379 return (yield* progress.environmentForThread(threadId)).T3_SCRIPT_PROGRESS_ENDPOINT!;
380 }).pipe(Effect.provide(testLayer));
381 const code = yield* Effect.callback<string | undefined>((resume) => {
382 const socket = NodeNet.connect({ host: "127.0.0.1", port: Number(endpoint.split(":")[1]) });
383 socket.once("connect", () => resume(Effect.die("Progress listener outlived its scope")));
384 socket.once("error", (error: NodeJS.ErrnoException) => resume(Effect.succeed(error.code)));
385 return Effect.sync(() => socket.destroy());
386 });
387 expect(code).toBe("ECONNREFUSED");
388 }),
389);
390
391it.effect(
392 "replaces a reconnected root while preserving another producer with the same PID and another thread",
393 () =>
394 Effect.gen(function* () {
395 const progress = yield* ScriptProgress.ScriptProgress;
396 const env = yield* progress.environmentForThread(threadId);
397 const otherEnv = yield* progress.environmentForThread(otherThreadId);
398 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
399 const pull = Queue.take(queue);
400 yield* pull;
401 const original = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
402 yield* write(original, {
403 type: "hello",
404 token: env.T3_SCRIPT_PROGRESS_TOKEN,
405 producerId: "producer-a",
406 pid: 42,
407 });
408 yield* write(original, {
409 type: "progress",
410 streamId: "root",
411 event: [-1, { k: 1, t: "Original", v: 7 }, [1]],
412 });
413 const originalEvent = yield* until(
414 pull,
415 (event) => event.type === "progress" && event.summary[0]?.text === "Original",
416 );
417 if (originalEvent.type !== "progress") throw new Error("Expected original progress");
418
419 const worker = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
420 yield* write(worker, {
421 type: "hello",
422 token: env.T3_SCRIPT_PROGRESS_TOKEN,
423 producerId: "producer-b",
424 pid: 42,
425 });
426 yield* write(worker, {
427 type: "progress",
428 streamId: "root",
429 event: [-1, { k: 1, t: "Worker" }, [1]],
430 });
431 yield* until(
432 pull,
433 (event) => event.type === "progress" && event.summary[0]?.text === "Worker",
434 );
435 const otherQueue = yield* Stream.toQueue(progress.subscribe(otherThreadId), {
436 capacity: "unbounded",
437 });
438 const otherPull = Queue.take(otherQueue);
439 yield* otherPull;
440 const other = yield* connect(otherEnv.T3_SCRIPT_PROGRESS_ENDPOINT!);
441 yield* write(other, {
442 type: "hello",
443 token: otherEnv.T3_SCRIPT_PROGRESS_TOKEN,
444 producerId: "producer-a",
445 pid: 42,
446 });
447 yield* write(other, {
448 type: "progress",
449 streamId: "root",
450 event: [-1, { k: 1, t: "Other thread" }, [1]],
451 });
452 yield* until(
453 otherPull,
454 (event) => event.type === "progress" && event.summary[0]?.text === "Other thread",
455 );
456
457 yield* Effect.sync(() => original.destroy());
458 yield* until(
459 pull,
460 (event) =>
461 event.type === "upsert" &&
462 event.stream.id === originalEvent.id &&
463 event.stream.status === "disconnected",
464 );
465 const reconnected = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
466 yield* write(reconnected, {
467 type: "hello",
468 token: env.T3_SCRIPT_PROGRESS_TOKEN,
469 producerId: "producer-a",
470 pid: 42,
471 });
472 yield* write(reconnected, {
473 type: "progress",
474 streamId: "root",
475 event: [-1, { k: 1, t: "Reconnected", v: 8 }, [1]],
476 });
477 yield* until(
478 pull,
479 (event) => event.type === "progress" && event.summary[0]?.text === "Reconnected",
480 );
481 const late = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
482 const snapshot = yield* Queue.take(late);
483 if (snapshot.type !== "snapshot") throw new Error("Expected snapshot");
484 expect(snapshot.streams).toHaveLength(2);
485 expect(snapshot.streams.every((stream) => stream.status === "connected")).toBe(true);
486 expect(snapshot.streams.map((stream) => stream.summary[0]?.text)).toEqual([
487 "Worker",
488 "Reconnected",
489 ]);
490 expect(snapshot.streams.some((stream) => stream.id === originalEvent.id)).toBe(false);
491 const otherLate = yield* Stream.toQueue(progress.subscribe(otherThreadId), {
492 capacity: "unbounded",
493 });
494 expect(yield* Queue.take(otherLate)).toMatchObject({
495 type: "snapshot",
496 streams: [{ status: "connected", summary: [{ text: "Other thread" }] }],
497 });
498 }).pipe(Effect.scoped, Effect.provide(testLayer)),
499);
500
501it.effect("uses the library visibility rule for native summaries", () =>
502 Effect.gen(function* () {
503 const progress = yield* ScriptProgress.ScriptProgress;
504 const env = yield* progress.environmentForThread(threadId);
505 const queue = yield* Stream.toQueue(progress.subscribe(threadId), { capacity: "unbounded" });
506 const pull = Queue.take(queue);
507 yield* pull;
508 const socket = yield* connect(env.T3_SCRIPT_PROGRESS_ENDPOINT!);
509 yield* write(socket, {
510 type: "hello",
511 token: env.T3_SCRIPT_PROGRESS_TOKEN,
512 producerId: "visibility",
513 pid: 45,
514 });
515 yield* write(socket, {
516 type: "progress",
517 streamId: "root",
518 event: [
519 -1,
520 { k: 1, t: "Empty group", p: true },
521 { k: 2, t: "Hidden", h: true },
522 { k: 3, t: "Visible", v: 2, e: 5 },
523 [1, 2, 3],
524 ],
525 });
526 const event = yield* until(
527 pull,
528 (event) => event.type === "progress" && event.summary.length > 0,
529 );
530 expect(event).toMatchObject({
531 type: "progress",
532 summary: [{ text: "Visible", value: 2, total: 5 }],
533 });
534 if (event.type !== "progress") throw new Error("Expected progress");
535 expect(event.summary).toHaveLength(1);
536 }).pipe(Effect.scoped, Effect.provide(testLayer)),
537);