diff --git a/packages/ssh/src/tunnel.test.ts b/packages/ssh/src/tunnel.test.ts index 7e7a5a54276..c2cbcc82f52 100644 --- a/packages/ssh/src/tunnel.test.ts +++ b/packages/ssh/src/tunnel.test.ts @@ -1,6 +1,8 @@ import { assert, describe, it } from "@effect/vitest"; +import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient"; 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"; @@ -77,6 +79,25 @@ const testHttpClient = HttpClient.make((request) => const hangingHttpClient = HttpClient.make(() => Effect.never); +const DELAYED_HTTP_FORWARD_SCRIPT = ` +const http = require("node:http"); +const port = Number(process.argv[1]); +const responseDelayMs = Number(process.argv[2]); +const server = http.createServer((_request, response) => { + setTimeout(() => { + response.writeHead(200, { "content-type": "text/plain" }); + response.end("ready"); + }, responseDelayMs); +}); +const shutdown = () => { + server.closeAllConnections?.(); + server.close(() => process.exit(0)); +}; +process.on("SIGINT", shutdown); +process.on("SIGTERM", shutdown); +server.listen(port, "127.0.0.1", () => process.stdout.write("listening\\n")); +`; + const testNetService = NetService.NetService.of({ canListenOnHost: () => Effect.succeed(true), isPortAvailableOnLoopback: () => Effect.succeed(true), @@ -392,4 +413,226 @@ describe("ssh tunnel scripts", () => { assert.equal(tunnelKillCount, 1); }).pipe(Effect.provide(layer), Effect.scoped); }); + + it.effect("keeps one authoritative tunnel across concurrent ensures", () => { + let tunnelSpawnCount = 0; + const spawner = ChildProcessSpawner.make((command) => + Effect.sync(() => { + const args = commandArgs(command); + if (args.includes("-N")) { + tunnelSpawnCount += 1; + return makeRunningProcess(() => undefined); + } + return makeSuccessfulProcess('{"remotePort":3773,"serverKind":"external"}\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 environments = yield* Effect.all( + Array.from({ length: 20 }, () => manager.ensureEnvironment(target)), + { concurrency: "unbounded" }, + ); + assert.equal(tunnelSpawnCount, 1); + assert.equal(new Set(environments.map((environment) => environment.httpBaseUrl)).size, 1); + }).pipe(Effect.provide(layer), Effect.scoped); + }); + + it.effect("waits for an in-flight ensure before disconnecting its tunnel", () => { + let activeTunnels = 0; + let tunnelKillCount = 0; + let remoteStopCount = 0; + const launchStarted = Deferred.makeUnsafe(); + const releaseLaunch = Deferred.makeUnsafe(); + const spawner = ChildProcessSpawner.make((command) => { + const args = commandArgs(command); + if (args.includes("-N")) { + return Effect.sync(() => { + activeTunnels += 1; + return makeRunningProcess(() => { + activeTunnels -= 1; + tunnelKillCount += 1; + }); + }); + } + if (args.includes("sh") && args.includes("--")) { + return Deferred.succeed(launchStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseLaunch)), + Effect.as(makeSuccessfulProcess('{"remotePort":3773,"serverKind":"external"}\n')), + ); + } + if (args.includes("sh")) { + remoteStopCount += 1; + return Effect.succeed(makeSuccessfulProcess('{"stopped":true}\n')); + } + return Effect.succeed(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 ensureFiber = yield* Effect.forkChild(manager.ensureEnvironment(target)); + yield* Deferred.await(launchStarted); + const disconnectFiber = yield* Effect.forkChild(manager.disconnectEnvironment(target)); + yield* Effect.yieldNow; + yield* Deferred.succeed(releaseLaunch, undefined); + yield* Fiber.join(ensureFiber); + yield* Fiber.join(disconnectFiber); + + assert.equal(activeTunnels, 0); + assert.equal(tunnelKillCount, 1); + assert.equal(remoteStopCount, 0); + }).pipe(Effect.provide(layer), Effect.scoped); + }); + + it.effect("reuses and reaps a real delayed local-forward process", () => + Effect.gen(function* () { + const platformSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const tunnelChildren: Array = []; + const localPorts: Array = []; + const spawner = ChildProcessSpawner.make((command) => { + const args = commandArgs(command); + if (!args.includes("-N")) { + return Effect.succeed( + makeSuccessfulProcess('{"remotePort":3773,"serverKind":"external"}\n'), + ); + } + const forwardIndex = args.indexOf("-L"); + const localPort = Number(args[forwardIndex + 1]?.split(":")[0]); + return platformSpawner + .spawn( + ChildProcess.make(process.execPath, [ + "-e", + DELAYED_HTTP_FORWARD_SCRIPT, + String(localPort), + "2600", + ]), + ) + .pipe( + Effect.tap((child) => + Effect.sync(() => { + tunnelChildren.push(child); + localPorts.push(localPort); + }), + ), + Effect.flatMap((child) => + child.stdout.pipe(Stream.decodeText(), Stream.runHead, Effect.as(child)), + ), + ); + }); + const layer = Layer.mergeAll( + NodeServices.layer, + NodeHttpClient.layerUndici, + NetService.layer, + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + SshPasswordPrompt.disabledLayer, + SshEnvironmentManager.layer(), + ); + const target = { + alias: "devbox", + hostname: "devbox.example.com", + username: "julius", + port: 2222, + } as const; + + yield* Effect.gen(function* () { + const manager = yield* SshEnvironmentManager; + const first = yield* manager.ensureEnvironment(target); + const second = yield* manager.ensureEnvironment(target); + + assert.equal(tunnelChildren.length, 1); + assert.equal(second.httpBaseUrl, first.httpBaseUrl); + + yield* manager.disconnectEnvironment(target); + + assert.deepEqual(yield* Effect.forEach(tunnelChildren, (child) => child.isRunning), [ + false, + ]); + const net = yield* NetService.NetService; + assert.isTrue(yield* net.canListenOnHost(localPorts[0]!, "127.0.0.1")); + }).pipe(Effect.provide(layer), Effect.scoped); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("does not grow tunnels or stop an external backend across 100 reconnects", () => { + let activeTunnels = 0; + let maximumActiveTunnels = 0; + let tunnelSpawnCount = 0; + let tunnelKillCount = 0; + let remoteStopCount = 0; + const spawner = ChildProcessSpawner.make((command) => + Effect.sync(() => { + const args = commandArgs(command); + if (args.includes("-N")) { + activeTunnels += 1; + maximumActiveTunnels = Math.max(maximumActiveTunnels, activeTunnels); + tunnelSpawnCount += 1; + return makeRunningProcess(() => { + activeTunnels -= 1; + tunnelKillCount += 1; + }); + } + if (args.includes("sh") && !args.includes("--")) { + remoteStopCount += 1; + return makeSuccessfulProcess('{"stopped":true}\n'); + } + return makeSuccessfulProcess('{"remotePort":3773,"serverKind":"external"}\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; + for (let reconnect = 0; reconnect < 100; reconnect += 1) { + yield* manager.ensureEnvironment(target); + assert.equal(activeTunnels, 1); + yield* manager.disconnectEnvironment(target); + assert.equal(activeTunnels, 0); + } + assert.equal(tunnelSpawnCount, 100); + assert.equal(tunnelKillCount, 100); + assert.equal(maximumActiveTunnels, 1); + assert.equal(remoteStopCount, 0); + }).pipe(Effect.provide(layer), Effect.scoped); + }); }); diff --git a/packages/ssh/src/tunnel.ts b/packages/ssh/src/tunnel.ts index 016d5e9a854..f9a48cba6dd 100644 --- a/packages/ssh/src/tunnel.ts +++ b/packages/ssh/src/tunnel.ts @@ -52,7 +52,8 @@ import { export const DEFAULT_REMOTE_PORT = 3773; const REMOTE_PORT_SCAN_WINDOW = 200; const SSH_READY_TIMEOUT_MS = 20_000; -const SSH_READY_PROBE_TIMEOUT_MS = 1_000; +const SSH_READY_PROBE_TIMEOUT_MS = 5_000; +const SSH_EXISTING_TUNNEL_READY_TIMEOUT_MS = 15_000; const TUNNEL_SHUTDOWN_TIMEOUT_MS = 2_000; const REMOTE_READY_TIMEOUT_MS = 15_000; const REMOTE_REUSE_READY_TIMEOUT_MS = 2_000; @@ -97,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, @@ -1168,18 +1160,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( @@ -1371,29 +1351,33 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma }); tunnels.delete(tunnelEntry.key); const authSecret = authSecrets.get(tunnelEntry.key) ?? null; + const stopOwnedRemoteServer = + tunnelEntry.remoteServerKind === "external" + ? Effect.void + : 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.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), - ), + stopOwnedRemoteServer, ], { concurrency: "unbounded" }, ).pipe(Effect.ignore); @@ -1429,7 +1413,10 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma remotePort: entry.remotePort, }); const readinessExit = yield* Effect.exit( - waitForHttpReady({ baseUrl: entry.httpBaseUrl, timeoutMs: 2_000 }), + waitForHttpReady({ + baseUrl: entry.httpBaseUrl, + timeoutMs: SSH_EXISTING_TUNNEL_READY_TIMEOUT_MS, + }), ); if (Exit.isSuccess(readinessExit)) { yield* Effect.logDebug("ssh.environment.tunnel.reused", { @@ -1448,7 +1435,6 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma cause: readinessExit.cause, }); yield* closeTunnelEntry(entry); - yield* cancelPendingTunnelEntry(key, resolvedTarget); entry = null; } @@ -1561,17 +1547,24 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma ...(target.port !== null ? { port: target.port } : {}), }; const key = targetConnectionKey(resolvedTarget); - const entry = tunnels.get(key) ?? null; + const pending = pendingTunnelEntries.get(key) ?? null; yield* Effect.logDebug("ssh.environment.disconnect.targetResolved", { ...sshTargetLogFields(resolvedTarget), key, - hasTunnel: entry !== null, - hasPendingTunnel: pendingTunnelEntries.has(key), + hasTunnel: tunnels.has(key), + hasPendingTunnel: pending !== null, }); + if (pending !== null) { + yield* Effect.logDebug("ssh.environment.disconnect.pending.await", { + ...sshTargetLogFields(resolvedTarget), + key, + }); + yield* Effect.exit(Deferred.await(pending)); + } + const entry = tunnels.get(key) ?? null; if (entry !== null) { yield* closeTunnelEntry(entry); } - yield* cancelPendingTunnelEntry(key, resolvedTarget); if (entry === null) { yield* runWithSshAuth({ key,