diff --git a/docs/internals/remote.md b/docs/internals/remote.md index 75a2f5ff8404..8f4bd360f95c 100644 --- a/docs/internals/remote.md +++ b/docs/internals/remote.md @@ -47,7 +47,9 @@ authorization. The desktop app can start or reuse Akeru Bot on an SSH host and open a local port forward. The remote host owns all server state. The desktop stores an `SshConnectionTarget` and reconnects -through the forwarded endpoint. +through the forwarded endpoint. Managed remote servers exec the resolved CLI without an npm wrapper +so the recorded PID is the server itself. A stop that does not confirm exit keeps ownership files +and reports failure; reconnect waits for that stop on the same target. SSH is a desktop capability because it requires local process and SSH access. Web and mobile can use a directly reachable server after normal pairing. diff --git a/packages/ssh/src/runnerProcess.test.ts b/packages/ssh/src/runnerProcess.test.ts new file mode 100644 index 000000000000..c323c42b7c90 --- /dev/null +++ b/packages/ssh/src/runnerProcess.test.ts @@ -0,0 +1,430 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import * as NodeNet from "node:net"; + +import { buildRemoteStopScript, buildRemoteT3RunnerScript } from "./tunnel.ts"; + +const Started = Schema.Struct({ + pid: Schema.Number, + port: Schema.Number, + args: Schema.Array(Schema.String), +}); +const decodeStarted = Schema.decodeUnknownSync(Schema.fromJsonString(Started)); + +describe.skipIf(HostProcessPlatform.defaultValue() === "win32")( + "remote runner process ownership", + () => { + it.live.each(["npx", "npm"] as const)( + "keeps the server PID and graceful shutdown through the %s fallback", + (packageManager) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fixture = yield* fs.makeTempDirectoryScoped({ prefix: "akeru-runner-" }); + const bin = path.join(fixture, "bin"); + const cliPath = path.join(fixture, "installed cli.mjs"); + const callsPath = path.join(fixture, "package-manager-calls.jsonl"); + const packageSpec = "akeru-bot@0.0.35"; + yield* fs.makeDirectory(bin); + yield* fs.symlink(process.execPath, path.join(bin, "node")); + yield* fs.writeFileString( + cliPath, + `#!/usr/bin/env node +import * as net from "node:net"; +const server = net.createServer((socket) => { + socket.end(); + server.close(); +}); +process.on("SIGTERM", () => server.close(() => { + process.stdout.write("graceful shutdown\\n"); +})); +server.listen(Number(process.env.T3_TEST_PORT ?? 0), "127.0.0.1", () => { + process.stdout.write(JSON.stringify({ + pid: process.pid, + port: server.address().port, + args: process.argv.slice(2), + }) + "\\n"); +}); +`, + ); + yield* fs.chmod(cliPath, 0o700); + yield* fs.writeFileString( + path.join(bin, packageManager), + `#!/usr/bin/env node +const fs = require("node:fs"); +const childProcess = require("node:child_process"); +const args = process.argv.slice(2); +fs.appendFileSync(process.env.T3_TEST_CALLS, JSON.stringify(args) + "\\n"); +if (args.includes("--package")) { + process.stdout.write(process.env.T3_TEST_CLI + "\\n"); +} else { + const child = childProcess.spawn(process.execPath, [process.env.T3_TEST_CLI, ...args], { stdio: "inherit" }); + child.once("exit", (code) => { process.exitCode = code ?? 1; }); +} +`, + ); + yield* fs.chmod(path.join(bin, packageManager), 0o700); + + const runServer = (port = 0) => + Effect.gen(function* () { + const child = yield* spawner.spawn( + ChildProcess.make("/bin/sh", ["-s", "--", "serve", "a path with spaces"], { + cwd: fixture, + env: { + PATH: bin, + T3_TEST_CLI: cliPath, + T3_TEST_CALLS: callsPath, + T3_TEST_PORT: String(port), + }, + detached: false, + stdin: Stream.make( + new TextEncoder().encode(buildRemoteT3RunnerScript({ packageSpec })), + ), + }), + ); + const ready = yield* Deferred.make(); + const stdout: string[] = []; + const output = yield* child.stdout.pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.runForEach((line) => + Effect.gen(function* () { + stdout.push(line); + if (stdout.length === 1) { + yield* Deferred.succeed(ready, decodeStarted(line)); + } + }), + ), + Effect.forkScoped, + ); + const stderr = yield* child.stderr.pipe( + Stream.decodeText(), + Stream.mkString, + Effect.forkScoped, + ); + const receipt = yield* Effect.raceFirst( + Deferred.await(ready), + Fiber.join(output).pipe( + Effect.flatMap(() => Fiber.join(stderr)), + Effect.flatMap((message) => + Effect.die(new Error(`Runner exited before listening: ${message}`)), + ), + ), + ); + // A failed PID assertion must still close the owned fixture server, including an npm child. + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + if (yield* child.isRunning) { + yield* Effect.callback((resume) => { + const connection = NodeNet.connect(receipt.port, "127.0.0.1"); + connection.on("error", () => undefined); + connection.once("close", () => resume(Effect.void)); + return Effect.sync(() => connection.destroy()); + }); + yield* child.exitCode; + } + }).pipe(Effect.orDie), + ); + assert.equal(receipt.pid, child.pid); + assert.deepEqual(receipt.args, ["serve", "a path with spaces"]); + yield* child.kill({ killSignal: "SIGTERM" }); + assert.equal(yield* child.exitCode, 0); + yield* Fiber.join(output); + assert.include(stdout, "graceful shutdown"); + return receipt.port; + }).pipe(Effect.scoped); + + const port = yield* runServer(); + assert.equal(yield* runServer(port), port); + const calls = (yield* fs.readFileString(callsPath)) + .trim() + .split("\n") + .map((line) => JSON.parse(line)); + const expectedCall = [ + ...(packageManager === "npm" ? ["exec"] : []), + "--yes", + "--package", + packageSpec, + "--", + "sh", + "-c", + "command -v akeru", + ]; + assert.deepEqual(calls, [expectedCall, expectedCall]); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + }, +); + +describe.skipIf(HostProcessPlatform.defaultValue() === "win32")( + "remote stop process ownership", + () => { + it.live.each(["graceful", "timeout", "external"] as const)( + "confirms the stop result for a %s server", + (mode) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fixture = yield* fs.makeTempDirectoryScoped({ prefix: "akeru-stop-" }); + const signalPath = path.join(fixture, "signals"); + const child = yield* spawner.spawn( + ChildProcess.make( + process.execPath, + [ + "--input-type=module", + "-e", + `import * as fs from "node:fs"; +import * as net from "node:net"; +const server = net.createServer((socket) => socket.end()); +let signals = 0; +process.on("SIGTERM", () => { + fs.writeFileSync(process.argv[2], String(++signals)); + if (process.argv[1] !== "timeout" || signals > 1) server.close(); +}); +server.listen(0, "127.0.0.1", () => { + process.stdout.write(JSON.stringify({ pid: process.pid, port: server.address().port, args: [] }) + "\\n"); +}); +`, + mode, + signalPath, + ], + { cwd: fixture, detached: false }, + ), + ); + // A failed assertion must still stop this captured fixture process. + yield* Effect.addFinalizer(() => + child.kill({ killSignal: "SIGKILL" }).pipe(Effect.ignore), + ); + const started = decodeStarted( + yield* child.stdout.pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.take(1), + Stream.mkString, + ), + ); + assert.equal(started.pid, child.pid); + const savedState = { + pid: `${child.pid}\n`, + port: `${started.port}\n`, + managed: mode === "external" ? "external\n" : "managed\n", + }; + for (const [name, contents] of Object.entries(savedState)) { + yield* fs.writeFileString(path.join(fixture, name), contents); + } + const script = buildRemoteStopScript({ + alias: "fixture", + hostname: "fixture", + username: null, + port: null, + }); + // Redirect only the state directory. Never use the developer's SSH state. + const isolatedScript = script.replace( + /^STATE_DIR=.*$/mu, + 'STATE_DIR="$T3_TEST_STATE_DIR"', + ); + assert.notEqual(isolatedScript, script); + const runStop = Effect.fn("test.remoteStop")(function* () { + const stop = yield* spawner.spawn( + ChildProcess.make("/bin/sh", ["-s"], { + cwd: fixture, + env: { T3_TEST_STATE_DIR: fixture }, + stdin: Stream.make(new TextEncoder().encode(isolatedScript)), + }), + ); + return yield* Effect.all( + { + stdout: stop.stdout.pipe(Stream.decodeText(), Stream.mkString), + stderr: stop.stderr.pipe(Stream.decodeText(), Stream.mkString), + exitCode: stop.exitCode, + }, + { concurrency: "unbounded" }, + ); + }, Effect.scoped); + let result = yield* runStop(); + if (mode !== "graceful") { + assert.isTrue(yield* child.isRunning); + yield* Effect.callback((resume) => { + const connection = NodeNet.connect(started.port, "127.0.0.1"); + connection.once("error", (error) => resume(Effect.fail(error))); + connection.once("close", () => resume(Effect.void)); + return Effect.sync(() => connection.destroy()); + }); + } + if (mode === "timeout") { + assert.equal(result.exitCode, 1); + assert.equal(result.stdout, ""); + assert.include(result.stderr, "did not stop within 2 seconds"); + assert.equal(yield* fs.readFileString(signalPath), "1"); + for (const [name, contents] of Object.entries(savedState)) { + assert.equal(yield* fs.readFileString(path.join(fixture, name)), contents); + } + result = yield* runStop(); + } + assert.equal(result.exitCode, 0); + assert.equal(result.stdout, '{"stopped":true}\n'); + assert.equal(result.stderr, ""); + for (const name of Object.keys(savedState)) { + assert.isFalse(yield* fs.exists(path.join(fixture, name))); + } + if (mode === "external") { + assert.isFalse(yield* fs.exists(signalPath)); + } else { + assert.equal(yield* child.exitCode, 0); + assert.equal(yield* fs.readFileString(signalPath), mode === "timeout" ? "2" : "1"); + } + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + }, +); + +describe.skipIf(HostProcessPlatform.defaultValue() === "win32")( + "remote runner install diagnostics", + () => { + const decodeArguments = Schema.decodeUnknownSync( + Schema.fromJsonString(Schema.Array(Schema.String)), + ); + const cases = (["npx", "npm"] as const).flatMap((packageManager) => + ( + [ + "etarget", + "network", + "empty-success", + "success", + "failed-with-path", + "existing-cli", + "node-override", + ] as const + ).map((mode) => ({ packageManager, mode })), + ); + + it.live.each(cases)("handles $packageManager/$mode", ({ packageManager, mode }) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fixture = yield* fs.makeTempDirectoryScoped({ prefix: "akeru-runner-install-" }); + const bin = path.join(fixture, "bin"); + const cliPath = path.join(fixture, "installed cli.mjs"); + const callsPath = path.join(fixture, "installer-calls.jsonl"); + const packageSpec = "akeru-bot@0.0.39-nightly.20260905.1286"; + const args = ["serve", "a path with spaces"]; + yield* fs.makeDirectory(bin); + yield* fs.symlink(process.execPath, path.join(bin, "node")); + yield* fs.writeFileString( + cliPath, + `#!/usr/bin/env node +process.stdout.write(JSON.stringify(process.argv.slice(2)) + "\\n"); +`, + ); + yield* fs.chmod(cliPath, 0o700); + yield* fs.writeFileString(callsPath, ""); + yield* fs.writeFileString( + path.join(bin, packageManager), + `#!/usr/bin/env node +const fs = require("node:fs"); +fs.appendFileSync(process.env.T3_TEST_CALLS, JSON.stringify(process.argv.slice(2)) + "\\n"); +const mode = process.env.T3_TEST_MODE; +if (mode === "success" || mode === "failed-with-path") { + process.stdout.write(process.env.T3_TEST_CLI + "\\n"); +} +if (mode === "etarget" || mode === "failed-with-path") { + process.stderr.write("npm error code ETARGET\\nnpm error notarget No matching version found.\\n"); + process.exitCode = 42; +} else if (mode === "network") { + process.stderr.write("npm error code ENETUNREACH\\n"); + process.exitCode = 43; +} +`, + ); + yield* fs.chmod(path.join(bin, packageManager), 0o700); + if (mode === "existing-cli") yield* fs.symlink(cliPath, path.join(bin, "akeru")); + + const child = yield* spawner.spawn( + ChildProcess.make("/bin/sh", ["-s", "--", ...args], { + cwd: fixture, + extendEnv: false, + env: { + PATH: bin, + T3_TEST_MODE: mode, + T3_TEST_CLI: cliPath, + T3_TEST_CALLS: callsPath, + }, + stdin: Stream.make( + new TextEncoder().encode( + buildRemoteT3RunnerScript({ + packageSpec, + ...(mode === "node-override" ? { nodeScriptPath: cliPath } : {}), + }), + ), + ), + }), + ); + const { stdout, stderr, exitCode } = yield* Effect.all( + { + stdout: child.stdout.pipe(Stream.decodeText(), Stream.mkString), + stderr: child.stderr.pipe(Stream.decodeText(), Stream.mkString), + exitCode: child.exitCode, + }, + { concurrency: "unbounded" }, + ); + const installFailed = + mode === "etarget" || mode === "network" || mode === "failed-with-path"; + const missingExecutable = mode === "empty-success"; + assert.equal(exitCode, installFailed || missingExecutable ? 1 : 0); + if (installFailed || missingExecutable) { + assert.equal(stdout, ""); + } else { + assert.deepEqual(decodeArguments(stdout), args); + } + if (installFailed) { + const npmError = mode === "network" ? "ENETUNREACH" : "ETARGET"; + assert.include(stderr, `npm error code ${npmError}\n`); + assert.include(stderr, `Remote host could not install ${packageSpec}.`); + assert.notInclude(stderr, "Remote host installed"); + assert.notInclude(stderr, "Install a C toolchain"); + } else if (missingExecutable) { + assert.include(stderr, `Remote host installed ${packageSpec}`); + assert.include(stderr, "npm produced no akeru executable"); + assert.include(stderr, "Install a C toolchain"); + } else { + assert.equal(stderr, ""); + } + const expectedCall = [ + ...(packageManager === "npm" ? ["exec"] : []), + "--yes", + "--package", + packageSpec, + "--", + "sh", + "-c", + "command -v akeru", + ]; + const usesInstaller = mode !== "existing-cli" && mode !== "node-override"; + const calls = yield* fs.readFileString(callsPath); + if (usesInstaller) { + assert.deepEqual( + calls + .trim() + .split("\n") + .map((line) => decodeArguments(line)), + [expectedCall], + ); + } else { + assert.equal(calls, ""); + } + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + }, +); diff --git a/packages/ssh/src/tunnel.test.ts b/packages/ssh/src/tunnel.test.ts index 22754ec03ff0..86d5c14b961b 100644 --- a/packages/ssh/src/tunnel.test.ts +++ b/packages/ssh/src/tunnel.test.ts @@ -1,6 +1,7 @@ import { assert, describe, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as NetService from "@t3tools/shared/Net"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -13,6 +14,7 @@ import { HttpClient, HttpClientResponse } from "effect/unstable/http"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { SshPasswordPrompt } from "./auth.ts"; +import { SshCommandError } from "./errors.ts"; import { buildRemoteLaunchScript, buildRemotePairingScript, @@ -105,8 +107,7 @@ describe("ssh tunnel scripts", () => { assert.include(script, "T3_NODE_SCRIPT_PATH=''"); assert.include(script, 'exec akeru "$@"'); - assert.include(script, "exec npx --yes 'akeru-bot@latest' \"$@\""); - assert.include(script, "exec npm exec --yes 'akeru-bot@latest' -- \"$@\""); + assert.include(script, 'exec "$T3_CLI_PATH" "$@"'); assert.include(script, "could not install 'akeru-bot@latest'"); assert.include(script, "require_installed_akeru_cli npx --yes --package 'akeru-bot@latest'"); assert.include( @@ -144,8 +145,6 @@ describe("ssh tunnel scripts", () => { packageSpec: "t3@nightly; touch /tmp/t3-owned", }); - assert.include(script, "exec npx --yes 't3@nightly; touch /tmp/t3-owned' \"$@\""); - assert.include(script, "exec npm exec --yes 't3@nightly; touch /tmp/t3-owned' -- \"$@\""); assert.include( script, "require_installed_akeru_cli npx --yes --package 't3@nightly; touch /tmp/t3-owned'", @@ -390,63 +389,199 @@ describe("ssh tunnel scripts", () => { }).pipe(Effect.provide(processLayer)); }); - it.effect("closes the tunnel scope and starts fresh after disconnect", () => { - const spawnedCommands: Array> = []; - let tunnelKillCount = 0; - let stopCommandCount = 0; - const spawner = ChildProcessSpawner.make((command) => - Effect.sync(() => { - const args = commandArgs(command); - spawnedCommands.push(args); - if (args.includes("-N")) { - return makeRunningProcess(() => { - tunnelKillCount += 1; - }); - } - if (args.includes("sh") && args.includes("--")) { - return makeSuccessfulProcess('{"remotePort":3773}\n'); - } - if (args.includes("sh")) { - stopCommandCount += 1; - return makeSuccessfulProcess('{"stopped":true}\n'); + it.effect.each(["successful stop", "failed stop"] as const)( + "closes the tunnel scope and starts fresh after a %s", + (mode) => { + const spawnedCommands: Array> = []; + let tunnelKillCount = 0; + let stopCommandCount = 0; + const spawner = ChildProcessSpawner.make((command) => + Effect.sync(() => { + const args = commandArgs(command); + spawnedCommands.push(args); + if (args.includes("-N")) { + return makeRunningProcess(() => { + tunnelKillCount += 1; + }); + } + if (args.includes("sh") && args.includes("--")) { + return makeSuccessfulProcess('{"remotePort":3773}\n'); + } + if (args.includes("sh")) { + stopCommandCount += 1; + if (mode === "failed stop" && stopCommandCount === 1) { + return { + ...makeSuccessfulProcess(""), + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)), + stderr: Stream.make( + new TextEncoder().encode("Remote Akeru server did not stop within 2 seconds.\n"), + ), + }; + } + return makeSuccessfulProcess('{"stopped":true}\n'); + } + return makeSuccessfulProcess("\n"); + }), + ); + const layer = Layer.mergeAll( + NodeServices.layer, + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + Layer.succeed(HttpClient.HttpClient, testHttpClient), + Layer.succeed(NetService.NetService, testNetService), + SshPasswordPrompt.disabledLayer, + SshEnvironmentManager.layer(), + ); + const target = { + alias: "devbox", + hostname: "devbox.example.com", + username: "julius", + port: 2222, + } as const; + + return Effect.gen(function* () { + const manager = yield* SshEnvironmentManager; + + const first = yield* manager.ensureEnvironment(target); + assert.equal(first.httpBaseUrl, "http://127.0.0.1:41773/"); + const firstTunnelArgs = spawnedCommands.find((args) => args.includes("-N")); + assert.isDefined(firstTunnelArgs); + assert.include(firstTunnelArgs, "ControlMaster=no"); + assert.include(firstTunnelArgs, "ControlPath=none"); + assert.include(firstTunnelArgs, "ControlPersist=no"); + + const disconnected = yield* Effect.result(manager.disconnectEnvironment(target)); + if (mode === "failed stop") { + assert.isTrue(Result.isFailure(disconnected)); + if (Result.isFailure(disconnected)) { + assert.instanceOf(disconnected.failure, SshCommandError); + assert.equal( + disconnected.failure.message, + "Remote Akeru server did not stop within 2 seconds.", + ); + } + } else { + assert.isTrue(Result.isSuccess(disconnected)); } - return makeSuccessfulProcess("\n"); - }), - ); - const layer = Layer.mergeAll( - NodeServices.layer, - Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), - Layer.succeed(HttpClient.HttpClient, testHttpClient), - Layer.succeed(NetService.NetService, testNetService), - SshPasswordPrompt.disabledLayer, - SshEnvironmentManager.layer(), - ); - const target = { - alias: "devbox", - hostname: "devbox.example.com", - username: "julius", - port: 2222, - } as const; - - return Effect.gen(function* () { - const manager = yield* SshEnvironmentManager; + assert.equal(tunnelKillCount, 1); + assert.equal(stopCommandCount, 1); - const first = yield* manager.ensureEnvironment(target); - assert.equal(first.httpBaseUrl, "http://127.0.0.1:41773/"); - const firstTunnelArgs = spawnedCommands.find((args) => args.includes("-N")); - assert.isDefined(firstTunnelArgs); - assert.include(firstTunnelArgs, "ControlMaster=no"); - assert.include(firstTunnelArgs, "ControlPath=none"); - assert.include(firstTunnelArgs, "ControlPersist=no"); - - yield* manager.disconnectEnvironment(target); - assert.equal(tunnelKillCount, 1); - assert.equal(stopCommandCount, 1); + if (mode === "failed stop") { + yield* manager.disconnectEnvironment(target); + assert.equal(tunnelKillCount, 1); + assert.equal(stopCommandCount, 2); + } - yield* manager.ensureEnvironment(target); + yield* manager.ensureEnvironment(target); + + assert.equal(spawnedCommands.filter((args) => args.includes("-N")).length, 2); + assert.equal(tunnelKillCount, 1); + }).pipe( + Effect.provide(layer), + Effect.scoped, + Effect.andThen( + Effect.sync(() => { + assert.equal(tunnelKillCount, 2); + assert.equal(stopCommandCount, mode === "failed stop" ? 3 : 2); + }), + ), + ); + }, + ); - assert.equal(spawnedCommands.filter((args) => args.includes("-N")).length, 2); - assert.equal(tunnelKillCount, 1); - }).pipe(Effect.provide(layer), Effect.scoped); - }); + it.effect.each(["local tunnel", "remote server"] as const)( + "waits for %s shutdown before reconnecting the same target", + (stalledStep) => + Effect.gen(function* () { + const shutdownStarted = yield* Deferred.make(); + const finishShutdown = yield* Deferred.make(); + const pauseShutdown = Deferred.succeed(shutdownStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishShutdown)), + ); + let launches = 0; + let tunnels = 0; + let stops = 0; + let remoteRunning = false; + const target = { alias: "devbox", hostname: "devbox", username: null, port: null }; + const spawner = ChildProcessSpawner.make((command) => + Effect.sync(() => { + const args = commandArgs(command); + const isTarget = args.includes(target.alias); + if (args.includes("-G")) { + return makeSuccessfulProcess(""); + } + if (args.includes("-N")) { + const tunnel = makeRunningProcess(() => undefined); + if (isTarget && ++tunnels === 1 && stalledStep === "local tunnel") { + return { + ...tunnel, + kill: (options?: ChildProcess.KillOptions) => + pauseShutdown.pipe(Effect.andThen(tunnel.kill(options))), + }; + } + return tunnel; + } + if (args.includes("--")) { + if (isTarget) { + launches += 1; + remoteRunning = true; + } + return makeSuccessfulProcess('{"remotePort":3773}\n'); + } + const stop = makeSuccessfulProcess('{"stopped":true}\n'); + if (!isTarget) return stop; + const pause = ++stops === 1 && stalledStep === "remote server"; + return { + ...stop, + exitCode: (pause ? pauseShutdown : Effect.void).pipe( + Effect.andThen( + Effect.sync(() => { + remoteRunning = false; + return ChildProcessSpawner.ExitCode(0); + }), + ), + ), + }; + }), + ); + const layer = Layer.mergeAll( + NodeServices.layer, + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + Layer.succeed(HttpClient.HttpClient, testHttpClient), + Layer.succeed(NetService.NetService, testNetService), + SshPasswordPrompt.disabledLayer, + SshEnvironmentManager.layer(), + ); + yield* Effect.gen(function* () { + const manager = yield* SshEnvironmentManager; + yield* manager.ensureEnvironment(target); + const disconnect = yield* Effect.forkChild(manager.disconnectEnvironment(target)); + yield* Deferred.await(shutdownStarted); + const firstReconnect = yield* Effect.forkChild(manager.ensureEnvironment(target)); + const secondReconnect = yield* Effect.forkChild(manager.ensureEnvironment(target)); + + yield* manager.ensureEnvironment({ + alias: "other", + hostname: "other", + username: null, + port: null, + }); + yield* TestClock.adjust(Duration.zero); + const launchesBeforeShutdown = launches; + yield* Deferred.succeed(finishShutdown, undefined); + yield* Fiber.join(disconnect); + const first = yield* Fiber.join(firstReconnect); + const second = yield* Fiber.join(secondReconnect); + + assert.equal(launchesBeforeShutdown, 1); + assert.equal(launches, 2); + assert.equal(tunnels, 2); + assert.isTrue(remoteRunning); + assert.equal(first.httpBaseUrl, second.httpBaseUrl); + }).pipe( + Effect.ensuring(Deferred.succeed(finishShutdown, undefined)), + Effect.provide(layer), + Effect.scoped, + ); + }), + ); }); diff --git a/packages/ssh/src/tunnel.ts b/packages/ssh/src/tunnel.ts index da7a716e1125..d7d15bca4bdc 100644 --- a/packages/ssh/src/tunnel.ts +++ b/packages/ssh/src/tunnel.ts @@ -10,7 +10,6 @@ import * as NetService from "@t3tools/shared/Net"; import { extractJsonObject, fromLenientJson } from "@t3tools/shared/schemaJson"; import { satisfiesSemverRange } from "@t3tools/shared/semver"; import * as Context from "effect/Context"; -import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; @@ -18,6 +17,7 @@ import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { HttpClient } from "effect/unstable/http"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; @@ -98,15 +98,6 @@ type SshEnvironmentEffectError = | SshPasswordPromptError | NetService.NetError; -function makeSshTunnelCancelledError(target: DesktopSshEnvironmentTarget): SshCommandError { - return new SshCommandError({ - command: ["ssh"], - exitCode: null, - stderr: "", - message: `SSH environment connection was cancelled for ${target.alias || target.hostname}.`, - }); -} - function sshTargetLogFields(target: DesktopSshEnvironmentTarget) { return { alias: target.alias, @@ -433,20 +424,24 @@ fi # never becomes ready. Resolve the CLI once up front so that install failure is # reported here, with npm's own output on stderr. require_installed_akeru_cli() { - T3_CLI_PATH="$("$@" -- sh -c 'command -v akeru' || true)" + if ! T3_CLI_PATH="$("$@" -- sh -c 'command -v akeru')"; then + printf 'Remote host could not install %s. See npm output above for the cause.\\n' @@T3_PACKAGE_SPEC@@ >&2 + return 1 + fi if [ -n "$T3_CLI_PATH" ]; then return 0 fi printf 'Remote host installed %s but npm produced no akeru executable, which usually means a native dependency (node-pty) failed to build. Install a C toolchain on the remote host (Debian/Ubuntu: build-essential, Fedora/RHEL: gcc-c++ make, macOS: xcode-select --install) and try again.\\n' @@T3_PACKAGE_SPEC@@ >&2 return 1 } +# The launcher records this PID, so exec the CLI without an npm wrapper process. if command -v npx >/dev/null 2>&1; then require_installed_akeru_cli npx --yes --package @@T3_PACKAGE_SPEC@@ || exit 1 - exec npx --yes @@T3_PACKAGE_SPEC@@ "$@" + exec "$T3_CLI_PATH" "$@" fi if command -v npm >/dev/null 2>&1; then require_installed_akeru_cli npm exec --yes --package @@T3_PACKAGE_SPEC@@ || exit 1 - exec npm exec --yes @@T3_PACKAGE_SPEC@@ -- "$@" + exec "$T3_CLI_PATH" "$@" fi printf 'Remote host is missing the akeru CLI and could not install @@T3_PACKAGE_SPEC@@ because node/npm/npx are unavailable on PATH. Install Node or configure a supported version manager for non-interactive shells.\\n' >&2 exit 1 @@ -638,6 +633,10 @@ if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ] && kill -0 "$REMO WAIT_COUNT=$((WAIT_COUNT + 1)) sleep 0.1 done + if kill -0 "$REMOTE_PID" 2>/dev/null; then + printf 'Remote Akeru server with PID %s did not stop within 2 seconds. Its ownership files were kept.\\n' "$REMOTE_PID" >&2 + exit 1 + fi fi rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" printf '{"stopped":true}\\n' @@ -1172,12 +1171,22 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma ): Effect.fn.Return { const managerScope = yield* Scope.Scope; const tunnels = new Map(); - const pendingTunnelEntries = new Map< - string, - Deferred.Deferred - >(); + const targetLocks = new Map(); const authSecrets = new Map(); + // Keep one lock per target so reconnect cannot reuse a server while stop is pending. + const withTargetLock = Effect.fn("ssh/tunnel.withTargetLock")(function* ( + key: string, + effect: Effect.Effect, + ): Effect.fn.Return { + let lock = targetLocks.get(key); + if (lock === undefined) { + lock = Semaphore.makeUnsafe(1); + targetLocks.set(key, lock); + } + return yield* lock.withPermits(1)(effect); + }); + const closeTunnelEntry = Effect.fn("ssh/tunnel.closeTunnelEntry")(function* ( entry: SshTunnelEntry, ) { @@ -1196,18 +1205,6 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma }); }); - const cancelPendingTunnelEntry = Effect.fn("ssh/tunnel.cancelPendingTunnelEntry")(function* ( - key: string, - target: DesktopSshEnvironmentTarget, - ) { - const pending = pendingTunnelEntries.get(key); - if (!pending) { - return; - } - pendingTunnelEntries.delete(key); - yield* Deferred.fail(pending, makeSshTunnelCancelledError(target)).pipe(Effect.ignore); - }); - yield* Scope.addFinalizer( managerScope, Effect.sync(() => [...tunnels.values()]).pipe( @@ -1388,7 +1385,17 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma yield* Scope.addFinalizer( entryScope, Effect.gen(function* () { - if (tunnels.get(tunnelEntry.key) !== tunnelEntry) { + const stopRemote = tunnels.get(tunnelEntry.key) === tunnelEntry; + if (stopRemote) { + tunnels.delete(tunnelEntry.key); + } + yield* tunnelEntry.process + .kill({ + killSignal: "SIGTERM", + forceKillAfter: TUNNEL_SHUTDOWN_TIMEOUT_MS, + }) + .pipe(Effect.ignore); + if (!stopRemote) { return; } yield* Effect.logDebug("ssh.environment.tunnel.finalizer.start", { @@ -1397,34 +1404,24 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma localPort: tunnelEntry.localPort, remotePort: tunnelEntry.remotePort, }); - tunnels.delete(tunnelEntry.key); const authSecret = authSecrets.get(tunnelEntry.key) ?? null; - yield* Effect.all( - [ - tunnelEntry.process.kill({ - killSignal: "SIGTERM", - forceKillAfter: TUNNEL_SHUTDOWN_TIMEOUT_MS, - }), - stopRemoteServer( - tunnelEntry.target, - authSecret === null - ? { - batchMode: "yes", - interactiveAuth: false, - } - : { - authSecret, - batchMode: "no", - interactiveAuth: true, - }, - ).pipe( - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawnerService), - Effect.provideService(FileSystem.FileSystem, fileSystemService), - Effect.provideService(Path.Path, pathService), - ), - ], - { concurrency: "unbounded" }, - ).pipe(Effect.ignore); + yield* stopRemoteServer( + tunnelEntry.target, + authSecret === null + ? { + batchMode: "yes", + interactiveAuth: false, + } + : { + authSecret, + batchMode: "no", + interactiveAuth: true, + }, + ).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawnerService), + Effect.provideService(FileSystem.FileSystem, fileSystemService), + Effect.provideService(Path.Path, pathService), + ); yield* Effect.logDebug("ssh.environment.tunnel.finalizer.succeeded", { ...sshTargetLogFields(tunnelEntry.target), key: tunnelEntry.key, @@ -1447,7 +1444,7 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma resolvedTarget: DesktopSshEnvironmentTarget, runner?: RemoteT3RunnerOptions, ): Effect.fn.Return { - let entry = tunnels.get(key) ?? null; + const entry = tunnels.get(key) ?? null; if (entry !== null) { yield* Effect.logDebug("ssh.environment.tunnel.existing.check", { @@ -1476,22 +1473,8 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma cause: readinessExit.cause, }); yield* closeTunnelEntry(entry); - yield* cancelPendingTunnelEntry(key, resolvedTarget); - entry = null; } - const pending = pendingTunnelEntries.get(key); - if (pending) { - yield* Effect.logDebug("ssh.environment.tunnel.pending.await", { - ...sshTargetLogFields(resolvedTarget), - key, - }); - return yield* Deferred.await(pending); - } - - const deferred = yield* Deferred.make(); - pendingTunnelEntries.set(key, deferred); - return yield* createTunnelEntry({ key, resolvedTarget, @@ -1504,13 +1487,6 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma cause, }), ), - Effect.onExit((exit) => - Effect.sync(() => { - if (pendingTunnelEntries.get(key) === deferred) { - pendingTunnelEntries.delete(key); - } - }).pipe(Effect.andThen(Deferred.done(deferred, exit))), - ), ); }); @@ -1526,91 +1502,102 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma ...sshTargetLogFields(target), issuePairingToken: requestOptions?.issuePairingToken === true, }); - const baseResolved = yield* resolveSshTarget(target.alias || target.hostname); - const resolvedTarget: DesktopSshEnvironmentTarget = { - ...baseResolved, - ...(target.username !== null ? { username: target.username } : {}), - ...(target.port !== null ? { port: target.port } : {}), - }; - const key = targetConnectionKey(resolvedTarget); - yield* Effect.logDebug("ssh.environment.target.resolved", { - ...sshTargetLogFields(resolvedTarget), - key, - }); - const packageSpec = options.resolveCliPackageSpec?.(); - const runner = - options.resolveCliRunner === undefined - ? packageSpec === undefined - ? undefined - : { packageSpec } - : yield* options.resolveCliRunner; - yield* Effect.logDebug("ssh.environment.runner.resolved", { - ...sshTargetLogFields(resolvedTarget), - ...sshRunnerLogFields(runner), - key, - }); - const entry = yield* ensureTunnelEntry(key, resolvedTarget, runner); + return yield* withTargetLock( + targetConnectionKey(target), + Effect.gen(function* () { + const baseResolved = yield* resolveSshTarget(target.alias || target.hostname); + const resolvedTarget: DesktopSshEnvironmentTarget = { + ...baseResolved, + ...(target.username !== null ? { username: target.username } : {}), + ...(target.port !== null ? { port: target.port } : {}), + }; + const key = targetConnectionKey(resolvedTarget); + yield* Effect.logDebug("ssh.environment.target.resolved", { + ...sshTargetLogFields(resolvedTarget), + key, + }); + const packageSpec = options.resolveCliPackageSpec?.(); + const runner = + options.resolveCliRunner === undefined + ? packageSpec === undefined + ? undefined + : { packageSpec } + : yield* options.resolveCliRunner; + yield* Effect.logDebug("ssh.environment.runner.resolved", { + ...sshTargetLogFields(resolvedTarget), + ...sshRunnerLogFields(runner), + key, + }); + const entry = yield* ensureTunnelEntry(key, resolvedTarget, runner); + + const pairingResult = requestOptions?.issuePairingToken + ? yield* runWithSshAuth({ + key, + target: entry.target, + operation: (authOptions) => + issueRemotePairingToken(entry.target, authOptions, runner), + }) + : null; + const pairingToken = pairingResult?.credential ?? null; - const pairingResult = requestOptions?.issuePairingToken - ? yield* runWithSshAuth({ + yield* Effect.logInfo("ssh.environment.ensure.succeeded", { + ...sshTargetLogFields(entry.target), key, + localPort: entry.localPort, + remotePort: entry.remotePort, + remoteServerKind: entry.remoteServerKind, + issuedPairingToken: pairingToken !== null, + }); + return { target: entry.target, - operation: (authOptions) => issueRemotePairingToken(entry.target, authOptions, runner), - }) - : null; - const pairingToken = pairingResult?.credential ?? null; - - yield* Effect.logInfo("ssh.environment.ensure.succeeded", { - ...sshTargetLogFields(entry.target), - key, - localPort: entry.localPort, - remotePort: entry.remotePort, - remoteServerKind: entry.remoteServerKind, - issuedPairingToken: pairingToken !== null, - }); - return { - target: entry.target, - httpBaseUrl: entry.httpBaseUrl, - wsBaseUrl: entry.wsBaseUrl, - pairingToken, - remotePort: entry.remotePort, - ...(entry.remoteServerKind ? { remoteServerKind: entry.remoteServerKind } : {}), - }; + httpBaseUrl: entry.httpBaseUrl, + wsBaseUrl: entry.wsBaseUrl, + pairingToken, + remotePort: entry.remotePort, + ...(entry.remoteServerKind ? { remoteServerKind: entry.remoteServerKind } : {}), + }; + }), + ); }); const disconnectEnvironment = Effect.fn("ssh/tunnel.disconnectEnvironment")(function* ( target: DesktopSshEnvironmentTarget, ): Effect.fn.Return { yield* Effect.logInfo("ssh.environment.disconnect.start", sshTargetLogFields(target)); - const baseResolved = yield* resolveSshTarget(target.alias || target.hostname); - const resolvedTarget: DesktopSshEnvironmentTarget = { - ...baseResolved, - ...(target.username !== null ? { username: target.username } : {}), - ...(target.port !== null ? { port: target.port } : {}), - }; - const key = targetConnectionKey(resolvedTarget); - const entry = tunnels.get(key) ?? null; - yield* Effect.logDebug("ssh.environment.disconnect.targetResolved", { - ...sshTargetLogFields(resolvedTarget), - key, - hasTunnel: entry !== null, - hasPendingTunnel: pendingTunnelEntries.has(key), - }); - if (entry !== null) { - yield* closeTunnelEntry(entry); - } - yield* cancelPendingTunnelEntry(key, resolvedTarget); - if (entry === null) { - yield* runWithSshAuth({ - key, - target: resolvedTarget, - operation: (authOptions) => stopRemoteServer(resolvedTarget, authOptions), - }); - } - yield* Effect.logInfo("ssh.environment.disconnect.succeeded", { - ...sshTargetLogFields(resolvedTarget), - key, - }); + yield* withTargetLock( + targetConnectionKey(target), + Effect.gen(function* () { + const baseResolved = yield* resolveSshTarget(target.alias || target.hostname); + const resolvedTarget: DesktopSshEnvironmentTarget = { + ...baseResolved, + ...(target.username !== null ? { username: target.username } : {}), + ...(target.port !== null ? { port: target.port } : {}), + }; + const key = targetConnectionKey(resolvedTarget); + const entry = tunnels.get(key) ?? null; + yield* Effect.logDebug("ssh.environment.disconnect.targetResolved", { + ...sshTargetLogFields(resolvedTarget), + key, + hasTunnel: entry !== null, + }); + if (entry !== null) { + // Explicit disconnect owns the remote stop so its failure reaches the caller. + yield* Effect.gen(function* () { + tunnels.delete(key); + yield* closeTunnelEntry(entry); + }).pipe(Effect.uninterruptible); + } + yield* runWithSshAuth({ + key, + target: resolvedTarget, + operation: (authOptions) => stopRemoteServer(resolvedTarget, authOptions), + }); + yield* Effect.logInfo("ssh.environment.disconnect.succeeded", { + ...sshTargetLogFields(resolvedTarget), + key, + }); + }), + ); }); return SshEnvironmentManager.of({ ensureEnvironment, disconnectEnvironment });