1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import * as HostProcess from "@t3tools/shared/HostProcess";
4import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";
7import * as FileSystem from "effect/FileSystem";
8import * as Path from "effect/Path";
9import * as Schema from "effect/Schema";
10import * as Stream from "effect/Stream";
11import { ChildProcess, ChildProcessSpawner } from "effect/process";
12import * as NodeNet from "node:net";
13
14import { remoteStateKey } from "./command.ts";
15import { buildRemoteStopScript, buildRemoteT3RunnerScript } from "./tunnel.ts";
16
17const Started = Schema.Struct({
18 pid: Schema.Number,
19 port: Schema.Number,
20 args: Schema.Array(Schema.String),
21});
22const decodeStarted = Schema.decodeUnknownSync(Schema.fromJsonString(Started));
23
24describe.skipIf(HostProcess.Platform.defaultValue() === "win32")(
25 "remote runner process ownership",
26 () => {
27 it.live("keeps the server PID and graceful shutdown through the node-script runner", () =>
28 Effect.gen(function* () {
29 const fs = yield* FileSystem.FileSystem;
30 const path = yield* Path.Path;
31 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
32 const fixture = yield* fs.makeTempDirectoryScoped({ prefix: "t3-runner-" });
33 const bin = path.join(fixture, "bin");
34 const cliPath = path.join(fixture, "installed cli.mjs");
35 yield* fs.makeDirectory(bin);
36 yield* fs.symlink(process.execPath, path.join(bin, "node"));
37 yield* fs.writeFileString(
38 cliPath,
39 `#!/usr/bin/env node
40import * as net from "node:net";
41const server = net.createServer((socket) => {
42 socket.end();
43 server.close();
44});
45process.on("SIGTERM", () => server.close(() => {
46 process.stdout.write("graceful shutdown\\n");
47}));
48server.listen(Number(process.env.T3_TEST_PORT ?? 0), "127.0.0.1", () => {
49 process.stdout.write(JSON.stringify({
50 pid: process.pid,
51 port: server.address().port,
52 args: process.argv.slice(2),
53 }) + "\\n");
54});
55`,
56 );
57 yield* fs.chmod(cliPath, 0o700);
58
59 const runServer = (port = 0) =>
60 Effect.gen(function* () {
61 const child = yield* spawner.spawn(
62 ChildProcess.make("/bin/sh", ["-s", "--", "serve", "a path with spaces"], {
63 cwd: fixture,
64 env: {
65 PATH: bin,
66 T3_TEST_PORT: String(port),
67 },
68 detached: false,
69 stdin: Stream.make(
70 new TextEncoder().encode(buildRemoteT3RunnerScript({ nodeScriptPath: cliPath })),
71 ),
72 }),
73 );
74 const ready = yield* Deferred.make<typeof Started.Type>();
75 const stdout: string[] = [];
76 const output = yield* child.stdout.pipe(
77 Stream.decodeText(),
78 Stream.splitLines,
79 Stream.runForEach((line) =>
80 Effect.gen(function* () {
81 stdout.push(line);
82 if (stdout.length === 1) {
83 yield* Deferred.succeed(ready, decodeStarted(line));
84 }
85 }),
86 ),
87 Effect.forkScoped,
88 );
89 const stderr = yield* child.stderr.pipe(
90 Stream.decodeText(),
91 Stream.mkString,
92 Effect.forkScoped,
93 );
94 const receipt = yield* Effect.raceFirst(
95 Deferred.await(ready),
96 Fiber.join(output).pipe(
97 Effect.flatMap(() => Fiber.join(stderr)),
98 Effect.flatMap((message) =>
99 Effect.die(new Error(`Runner exited before listening: ${message}`)),
100 ),
101 ),
102 );
103 // A failed PID assertion must still close the owned fixture server.
104 yield* Effect.addFinalizer(() =>
105 Effect.gen(function* () {
106 if (yield* child.isRunning) {
107 yield* Effect.callback<void>((resume) => {
108 const connection = NodeNet.connect(receipt.port, "127.0.0.1");
109 connection.on("error", () => undefined);
110 connection.once("close", () => resume(Effect.void));
111 return Effect.sync(() => connection.destroy());
112 });
113 yield* child.exitCode;
114 }
115 }).pipe(Effect.orDie),
116 );
117 assert.equal(receipt.pid, child.pid);
118 assert.deepEqual(receipt.args, ["serve", "a path with spaces"]);
119 yield* child.kill({ killSignal: "SIGTERM" });
120 assert.equal(yield* child.exitCode, 0);
121 yield* Fiber.join(output);
122 assert.include(stdout, "graceful shutdown");
123 return receipt.port;
124 }).pipe(Effect.scoped);
125
126 const port = yield* runServer();
127 assert.equal(yield* runServer(port), port);
128 }).pipe(Effect.provide(NodeServices.layer), Effect.scoped),
129 );
130 },
131);
132
133describe.skipIf(HostProcess.Platform.defaultValue() === "win32")(
134 "remote stop process ownership",
135 () => {
136 it.live.each(["graceful", "timeout", "external"] as const)(
137 "confirms the stop result for a %s server",
138 (mode) =>
139 Effect.gen(function* () {
140 const fs = yield* FileSystem.FileSystem;
141 const path = yield* Path.Path;
142 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
143 const fixture = yield* fs.makeTempDirectoryScoped({ prefix: "t3-stop-" });
144 const signalPath = path.join(fixture, "signals");
145 const child = yield* spawner.spawn(
146 ChildProcess.make(
147 process.execPath,
148 [
149 "--input-type=module",
150 "-e",
151 `import * as fs from "node:fs";
152import * as net from "node:net";
153const server = net.createServer((socket) => socket.end());
154let signals = 0;
155process.on("SIGTERM", () => {
156 fs.writeFileSync(process.argv[2], String(++signals));
157 if (process.argv[1] !== "timeout" || signals > 1) server.close();
158});
159server.listen(0, "127.0.0.1", () => {
160 process.stdout.write(JSON.stringify({ pid: process.pid, port: server.address().port, args: [] }) + "\\n");
161});
162`,
163 mode,
164 signalPath,
165 ],
166 { cwd: fixture, detached: false },
167 ),
168 );
169 // A failed assertion must still stop this captured fixture process.
170 yield* Effect.addFinalizer(() =>
171 child.kill({ killSignal: "SIGKILL" }).pipe(Effect.ignore),
172 );
173 const started = decodeStarted(
174 yield* child.stdout.pipe(
175 Stream.decodeText(),
176 Stream.splitLines,
177 Stream.take(1),
178 Stream.mkString,
179 ),
180 );
181 assert.equal(started.pid, child.pid);
182 const savedState = {
183 pid: `${child.pid}\n`,
184 port: `${started.port}\n`,
185 managed: mode === "external" ? "external\n" : "managed\n",
186 };
187 for (const [name, contents] of Object.entries(savedState)) {
188 yield* fs.writeFileString(path.join(fixture, name), contents);
189 }
190 const script = buildRemoteStopScript(
191 yield* remoteStateKey({
192 alias: "fixture",
193 hostname: "fixture",
194 username: null,
195 port: null,
196 }),
197 );
198 // Redirect only the state directory. Never use the developer's SSH state.
199 const isolatedScript = script.replace(
200 /^STATE_DIR=.*$/mu,
201 'STATE_DIR="$T3_TEST_STATE_DIR"',
202 );
203 assert.notEqual(isolatedScript, script);
204 const runStop = Effect.fn("test.remoteStop")(function* () {
205 const stop = yield* spawner.spawn(
206 ChildProcess.make("/bin/sh", ["-s"], {
207 cwd: fixture,
208 env: { T3_TEST_STATE_DIR: fixture },
209 stdin: Stream.make(new TextEncoder().encode(isolatedScript)),
210 }),
211 );
212 return yield* Effect.all(
213 {
214 stdout: stop.stdout.pipe(Stream.decodeText(), Stream.mkString),
215 stderr: stop.stderr.pipe(Stream.decodeText(), Stream.mkString),
216 exitCode: stop.exitCode,
217 },
218 { concurrency: "unbounded" },
219 );
220 }, Effect.scoped);
221 let result = yield* runStop();
222 if (mode !== "graceful") {
223 assert.isTrue(yield* child.isRunning);
224 yield* Effect.callback<void, Error>((resume) => {
225 const connection = NodeNet.connect(started.port, "127.0.0.1");
226 connection.once("error", (error) => resume(Effect.fail(error)));
227 connection.once("close", () => resume(Effect.void));
228 return Effect.sync(() => connection.destroy());
229 });
230 }
231 if (mode === "timeout") {
232 assert.equal(result.exitCode, 1);
233 assert.equal(result.stdout, "");
234 assert.include(result.stderr, "did not stop within 2 seconds");
235 assert.equal(yield* fs.readFileString(signalPath), "1");
236 for (const [name, contents] of Object.entries(savedState)) {
237 assert.equal(yield* fs.readFileString(path.join(fixture, name)), contents);
238 }
239 result = yield* runStop();
240 }
241 assert.equal(result.exitCode, 0);
242 assert.equal(result.stdout, '{"stopped":true}\n');
243 assert.equal(result.stderr, "");
244 for (const name of Object.keys(savedState)) {
245 assert.isFalse(yield* fs.exists(path.join(fixture, name)));
246 }
247 if (mode === "external") {
248 assert.isFalse(yield* fs.exists(signalPath));
249 } else {
250 assert.equal(yield* child.exitCode, 0);
251 assert.equal(yield* fs.readFileString(signalPath), mode === "timeout" ? "2" : "1");
252 }
253 }).pipe(Effect.provide(NodeServices.layer), Effect.scoped),
254 );
255 },
256);