authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-06 02:15:49-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-06 02:15:49-07:00
log60e6fa30c3a8c1e17ba7fb80e5f85f44a0cd2693
treeb8c583c22c2346eef59c19bafea728e598cfdb73
parentf66cfe2219e3f255f573f6430f9c989c58ed517d
signature Signed by PGP key B5690EEEBB952194

fix(ssh): report remote stop failures without losing ownership (#10105)

3 files changed, 427 insertions(+), 177 deletions(-)

packages/ssh/src/runnerProcess.test.ts+124-1
......@@ -11,7 +11,7 @@ import * as Stream from "effect/Stream";
1111import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
1212import * as NodeNet from "node:net";
1313
14import { buildRemoteT3RunnerScript } from "./tunnel.ts";
14import { buildRemoteStopScript, buildRemoteT3RunnerScript } from "./tunnel.ts";
1515
1616const Started = Schema.Struct({
1717 pid: Schema.Number,
......@@ -166,6 +166,129 @@ if (args.includes("--package")) {
166166 },
167167);
168168
169describe.skipIf(HostProcessPlatform.defaultValue() === "win32")(
170 "remote stop process ownership",
171 () => {
172 it.live.each(["graceful", "timeout", "external"] as const)(
173 "confirms the stop result for a %s server",
174 (mode) =>
175 Effect.gen(function* () {
176 const fs = yield* FileSystem.FileSystem;
177 const path = yield* Path.Path;
178 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
179 const fixture = yield* fs.makeTempDirectoryScoped({ prefix: "t3-stop-" });
180 const signalPath = path.join(fixture, "signals");
181 const child = yield* spawner.spawn(
182 ChildProcess.make(
183 process.execPath,
184 [
185 "--input-type=module",
186 "-e",
187 `import * as fs from "node:fs";
188import * as net from "node:net";
189const server = net.createServer((socket) => socket.end());
190let signals = 0;
191process.on("SIGTERM", () => {
192 fs.writeFileSync(process.argv[2], String(++signals));
193 if (process.argv[1] !== "timeout" || signals > 1) server.close();
194});
195server.listen(0, "127.0.0.1", () => {
196 process.stdout.write(JSON.stringify({ pid: process.pid, port: server.address().port, args: [] }) + "\\n");
197});
198`,
199 mode,
200 signalPath,
201 ],
202 { cwd: fixture, detached: false },
203 ),
204 );
205 // A failed assertion must still stop this captured fixture process.
206 yield* Effect.addFinalizer(() =>
207 child.kill({ killSignal: "SIGKILL" }).pipe(Effect.ignore),
208 );
209 const started = decodeStarted(
210 yield* child.stdout.pipe(
211 Stream.decodeText(),
212 Stream.splitLines,
213 Stream.take(1),
214 Stream.mkString,
215 ),
216 );
217 assert.equal(started.pid, child.pid);
218 const savedState = {
219 pid: `${child.pid}\n`,
220 port: `${started.port}\n`,
221 managed: mode === "external" ? "external\n" : "managed\n",
222 };
223 for (const [name, contents] of Object.entries(savedState)) {
224 yield* fs.writeFileString(path.join(fixture, name), contents);
225 }
226 const script = buildRemoteStopScript({
227 alias: "fixture",
228 hostname: "fixture",
229 username: null,
230 port: null,
231 });
232 // Redirect only the state directory. Never use the developer's SSH state.
233 const isolatedScript = script.replace(
234 /^STATE_DIR=.*$/mu,
235 'STATE_DIR="$T3_TEST_STATE_DIR"',
236 );
237 assert.notEqual(isolatedScript, script);
238 const runStop = Effect.fn("test.remoteStop")(function* () {
239 const stop = yield* spawner.spawn(
240 ChildProcess.make("/bin/sh", ["-s"], {
241 cwd: fixture,
242 env: { T3_TEST_STATE_DIR: fixture },
243 stdin: Stream.make(new TextEncoder().encode(isolatedScript)),
244 }),
245 );
246 return yield* Effect.all(
247 {
248 stdout: stop.stdout.pipe(Stream.decodeText(), Stream.mkString),
249 stderr: stop.stderr.pipe(Stream.decodeText(), Stream.mkString),
250 exitCode: stop.exitCode,
251 },
252 { concurrency: "unbounded" },
253 );
254 }, Effect.scoped);
255 let result = yield* runStop();
256 if (mode !== "graceful") {
257 assert.isTrue(yield* child.isRunning);
258 yield* Effect.callback<void, Error>((resume) => {
259 const connection = NodeNet.connect(started.port, "127.0.0.1");
260 connection.once("error", (error) => resume(Effect.fail(error)));
261 connection.once("close", () => resume(Effect.void));
262 return Effect.sync(() => connection.destroy());
263 });
264 }
265 if (mode === "timeout") {
266 assert.equal(result.exitCode, 1);
267 assert.equal(result.stdout, "");
268 assert.include(result.stderr, "did not stop within 2 seconds");
269 assert.equal(yield* fs.readFileString(signalPath), "1");
270 for (const [name, contents] of Object.entries(savedState)) {
271 assert.equal(yield* fs.readFileString(path.join(fixture, name)), contents);
272 }
273 result = yield* runStop();
274 }
275 assert.equal(result.exitCode, 0);
276 assert.equal(result.stdout, '{"stopped":true}\n');
277 assert.equal(result.stderr, "");
278 for (const name of Object.keys(savedState)) {
279 assert.isFalse(yield* fs.exists(path.join(fixture, name)));
280 }
281 if (mode === "external") {
282 assert.isFalse(yield* fs.exists(signalPath));
283 } else {
284 assert.equal(yield* child.exitCode, 0);
285 assert.equal(yield* fs.readFileString(signalPath), mode === "timeout" ? "2" : "1");
286 }
287 }).pipe(Effect.provide(NodeServices.layer), Effect.scoped),
288 );
289 },
290);
291
169292describe.skipIf(HostProcessPlatform.defaultValue() === "win32")(
170293 "remote runner install diagnostics",
171294 () => {
packages/ssh/src/tunnel.test.ts+199-55
......@@ -1,6 +1,7 @@
11import { assert, describe, it } from "@effect/vitest";
22import * as NodeServices from "@effect/platform-node/NodeServices";
33import * as NetService from "@t3tools/shared/Net";
4import * as Deferred from "effect/Deferred";
45import * as Duration from "effect/Duration";
56import * as Effect from "effect/Effect";
67import * as Fiber from "effect/Fiber";
......@@ -13,6 +14,7 @@ import { HttpClient, HttpClientResponse } from "effect/unstable/http";
1314import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
1415
1516import { SshPasswordPrompt } from "./auth.ts";
17import { SshCommandError } from "./errors.ts";
1618import {
1719 buildRemoteLaunchScript,
1820 buildRemotePairingScript,
......@@ -384,63 +386,205 @@ describe("ssh tunnel scripts", () => {
384386 }).pipe(Effect.provide(processLayer));
385387 });
386388
387 it.effect("closes the tunnel scope and starts fresh after disconnect", () => {
388 const spawnedCommands: Array<ReadonlyArray<string>> = [];
389 let tunnelKillCount = 0;
390 let stopCommandCount = 0;
391 const spawner = ChildProcessSpawner.make((command) =>
392 Effect.sync(() => {
393 const args = commandArgs(command);
394 spawnedCommands.push(args);
395 if (args.includes("-N")) {
396 return makeRunningProcess(() => {
397 tunnelKillCount += 1;
398 });
399 }
400 if (args.includes("sh") && args.includes("--")) {
401 return makeSuccessfulProcess('{"remotePort":3773}\n');
402 }
403 if (args.includes("sh")) {
404 stopCommandCount += 1;
405 return makeSuccessfulProcess('{"stopped":true}\n');
389 it.effect.each(["successful stop", "failed stop"] as const)(
390 "closes the tunnel scope and starts fresh after a %s",
391 (mode) => {
392 const spawnedCommands: Array<ReadonlyArray<string>> = [];
393 let tunnelKillCount = 0;
394 let stopCommandCount = 0;
395 const spawner = ChildProcessSpawner.make((command) =>
396 Effect.sync(() => {
397 const args = commandArgs(command);
398 spawnedCommands.push(args);
399 if (args.includes("-N")) {
400 return makeRunningProcess(() => {
401 tunnelKillCount += 1;
402 });
403 }
404 if (args.includes("sh") && args.includes("--")) {
405 return makeSuccessfulProcess('{"remotePort":3773}\n');
406 }
407 if (args.includes("sh")) {
408 stopCommandCount += 1;
409 if (mode === "failed stop" && stopCommandCount === 1) {
410 return {
411 ...makeSuccessfulProcess(""),
412 exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)),
413 stderr: Stream.make(
414 new TextEncoder().encode("Remote T3 server did not stop within 2 seconds.\n"),
415 ),
416 };
417 }
418 return makeSuccessfulProcess('{"stopped":true}\n');
419 }
420 return makeSuccessfulProcess("\n");
421 }),
422 );
423 const layer = Layer.mergeAll(
424 NodeServices.layer,
425 Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner),
426 Layer.succeed(HttpClient.HttpClient, testHttpClient),
427 Layer.succeed(NetService.NetService, testNetService),
428 SshPasswordPrompt.disabledLayer,
429 SshEnvironmentManager.layer(),
430 );
431 const target = {
432 alias: "devbox",
433 hostname: "devbox.example.com",
434 username: "julius",
435 port: 2222,
436 } as const;
437
438 return Effect.gen(function* () {
439 const manager = yield* SshEnvironmentManager;
440
441 const first = yield* manager.ensureEnvironment(target);
442 assert.equal(first.httpBaseUrl, "http://127.0.0.1:41773/");
443 const firstTunnelArgs = spawnedCommands.find((args) => args.includes("-N"));
444 assert.isDefined(firstTunnelArgs);
445 assert.include(firstTunnelArgs, "ControlMaster=no");
446 assert.include(firstTunnelArgs, "ControlPath=none");
447 assert.include(firstTunnelArgs, "ControlPersist=no");
448
449 const disconnected = yield* Effect.result(manager.disconnectEnvironment(target));
450 if (mode === "failed stop") {
451 assert.isTrue(Result.isFailure(disconnected));
452 if (Result.isFailure(disconnected)) {
453 assert.instanceOf(disconnected.failure, SshCommandError);
454 assert.equal(
455 disconnected.failure.message,
456 "Remote T3 server did not stop within 2 seconds.",
457 );
458 }
459 } else {
460 assert.isTrue(Result.isSuccess(disconnected));
406461 }
407 return makeSuccessfulProcess("\n");
408 }),
409 );
410 const layer = Layer.mergeAll(
411 NodeServices.layer,
412 Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner),
413 Layer.succeed(HttpClient.HttpClient, testHttpClient),
414 Layer.succeed(NetService.NetService, testNetService),
415 SshPasswordPrompt.disabledLayer,
416 SshEnvironmentManager.layer(),
417 );
418 const target = {
419 alias: "devbox",
420 hostname: "devbox.example.com",
421 username: "julius",
422 port: 2222,
423 } as const;
424
425 return Effect.gen(function* () {
426 const manager = yield* SshEnvironmentManager;
462 assert.equal(tunnelKillCount, 1);
463 assert.equal(stopCommandCount, 1);
427464
428 const first = yield* manager.ensureEnvironment(target);
429 assert.equal(first.httpBaseUrl, "http://127.0.0.1:41773/");
430 const firstTunnelArgs = spawnedCommands.find((args) => args.includes("-N"));
431 assert.isDefined(firstTunnelArgs);
432 assert.include(firstTunnelArgs, "ControlMaster=no");
433 assert.include(firstTunnelArgs, "ControlPath=none");
434 assert.include(firstTunnelArgs, "ControlPersist=no");
435
436 yield* manager.disconnectEnvironment(target);
437 assert.equal(tunnelKillCount, 1);
438 assert.equal(stopCommandCount, 1);
465 if (mode === "failed stop") {
466 yield* manager.disconnectEnvironment(target);
467 assert.equal(tunnelKillCount, 1);
468 assert.equal(stopCommandCount, 2);
469 }
439470
440 yield* manager.ensureEnvironment(target);
471 yield* manager.ensureEnvironment(target);
472
473 assert.equal(spawnedCommands.filter((args) => args.includes("-N")).length, 2);
474 assert.equal(tunnelKillCount, 1);
475 }).pipe(
476 Effect.provide(layer),
477 Effect.scoped,
478 Effect.andThen(
479 Effect.sync(() => {
480 assert.equal(tunnelKillCount, 2);
481 assert.equal(stopCommandCount, mode === "failed stop" ? 3 : 2);
482 }),
483 ),
484 );
485 },
486 );
441487
442 assert.equal(spawnedCommands.filter((args) => args.includes("-N")).length, 2);
443 assert.equal(tunnelKillCount, 1);
444 }).pipe(Effect.provide(layer), Effect.scoped);
445 });
488 it.effect.each(["local tunnel", "remote server"] as const)(
489 "waits for %s shutdown before reconnecting the same target",
490 (stalledStep) =>
491 Effect.gen(function* () {
492 const shutdownStarted = yield* Deferred.make<void>();
493 const finishShutdown = yield* Deferred.make<void>();
494 const reconnectsStarted = yield* Deferred.make<void>();
495 const pauseShutdown = Deferred.succeed(shutdownStarted, undefined).pipe(
496 Effect.andThen(Deferred.await(finishShutdown)),
497 );
498 let resolutions = 0;
499 let launches = 0;
500 let tunnels = 0;
501 let stops = 0;
502 let remoteRunning = false;
503 const target = { alias: "devbox", hostname: "devbox", username: null, port: null };
504 const spawner = ChildProcessSpawner.make((command) =>
505 Effect.gen(function* () {
506 const args = commandArgs(command);
507 const isTarget = args.includes(target.alias);
508 if (args.includes("-G")) {
509 if (isTarget && ++resolutions === 4) {
510 yield* Deferred.succeed(reconnectsStarted, undefined);
511 }
512 return makeSuccessfulProcess("");
513 }
514 if (args.includes("-N")) {
515 const tunnel = makeRunningProcess(() => undefined);
516 if (isTarget && ++tunnels === 1 && stalledStep === "local tunnel") {
517 return {
518 ...tunnel,
519 kill: (options?: ChildProcess.KillOptions) =>
520 pauseShutdown.pipe(Effect.andThen(tunnel.kill(options))),
521 };
522 }
523 return tunnel;
524 }
525 if (args.includes("--")) {
526 if (isTarget) {
527 launches += 1;
528 remoteRunning = true;
529 }
530 return makeSuccessfulProcess('{"remotePort":3773}\n');
531 }
532 const stop = makeSuccessfulProcess('{"stopped":true}\n');
533 if (!isTarget) return stop;
534 const pause = ++stops === 1 && stalledStep === "remote server";
535 return {
536 ...stop,
537 exitCode: (pause ? pauseShutdown : Effect.void).pipe(
538 Effect.andThen(
539 Effect.sync(() => {
540 remoteRunning = false;
541 return ChildProcessSpawner.ExitCode(0);
542 }),
543 ),
544 ),
545 };
546 }),
547 );
548 const layer = Layer.mergeAll(
549 NodeServices.layer,
550 Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner),
551 Layer.succeed(HttpClient.HttpClient, testHttpClient),
552 Layer.succeed(NetService.NetService, testNetService),
553 SshPasswordPrompt.disabledLayer,
554 SshEnvironmentManager.layer(),
555 );
556 yield* Effect.gen(function* () {
557 const manager = yield* SshEnvironmentManager;
558 yield* manager.ensureEnvironment(target);
559 const disconnect = yield* Effect.forkChild(manager.disconnectEnvironment(target));
560 yield* Deferred.await(shutdownStarted);
561 const firstReconnect = yield* Effect.forkChild(manager.ensureEnvironment(target));
562 const secondReconnect = yield* Effect.forkChild(manager.ensureEnvironment(target));
563 yield* Deferred.await(reconnectsStarted);
564
565 yield* manager.ensureEnvironment({
566 alias: "other",
567 hostname: "other",
568 username: null,
569 port: null,
570 });
571 yield* TestClock.adjust(Duration.zero);
572 const launchesBeforeShutdown = launches;
573 yield* Deferred.succeed(finishShutdown, undefined);
574 yield* Fiber.join(disconnect);
575 const first = yield* Fiber.join(firstReconnect);
576 const second = yield* Fiber.join(secondReconnect);
577
578 assert.equal(launchesBeforeShutdown, 1);
579 assert.equal(launches, 2);
580 assert.equal(tunnels, 2);
581 assert.isTrue(remoteRunning);
582 assert.equal(first.httpBaseUrl, second.httpBaseUrl);
583 }).pipe(
584 Effect.ensuring(Deferred.succeed(finishShutdown, undefined)),
585 Effect.provide(layer),
586 Effect.scoped,
587 );
588 }),
589 );
446590});
packages/ssh/src/tunnel.ts+104-121
......@@ -10,7 +10,6 @@ import * as NetService from "@t3tools/shared/Net";
1010import { extractJsonObject, fromLenientJson } from "@t3tools/shared/schemaJson";
1111import { satisfiesSemverRange } from "@t3tools/shared/semver";
1212import * as Context from "effect/Context";
13import * as Deferred from "effect/Deferred";
1413import * as Effect from "effect/Effect";
1514import * as Exit from "effect/Exit";
1615import * as FileSystem from "effect/FileSystem";
......@@ -18,6 +17,7 @@ import * as Layer from "effect/Layer";
1817import * as Path from "effect/Path";
1918import * as Schema from "effect/Schema";
2019import * as Scope from "effect/Scope";
20import * as Semaphore from "effect/Semaphore";
2121import * as Stream from "effect/Stream";
2222import { HttpClient } from "effect/unstable/http";
2323import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
......@@ -98,15 +98,6 @@ type SshEnvironmentEffectError =
9898 | SshPasswordPromptError
9999 | NetService.NetError;
100100
101function makeSshTunnelCancelledError(target: DesktopSshEnvironmentTarget): SshCommandError {
102 return new SshCommandError({
103 command: ["ssh"],
104 exitCode: null,
105 stderr: "",
106 message: `SSH environment connection was cancelled for ${target.alias || target.hostname}.`,
107 });
108}
109
110101function sshTargetLogFields(target: DesktopSshEnvironmentTarget) {
111102 return {
112103 alias: target.alias,
......@@ -642,6 +633,10 @@ if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ] && kill -0 "$REMO
642633 WAIT_COUNT=$((WAIT_COUNT + 1))
643634 sleep 0.1
644635 done
636 if kill -0 "$REMOTE_PID" 2>/dev/null; then
637 printf 'Remote T3 server with PID %s did not stop within 2 seconds. Its ownership files were kept.\\n' "$REMOTE_PID" >&2
638 exit 1
639 fi
645640fi
646641rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE"
647642printf '{"stopped":true}\\n'
......@@ -1176,12 +1171,22 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
11761171): Effect.fn.Return<SshEnvironmentManagerShape, never, Scope.Scope> {
11771172 const managerScope = yield* Scope.Scope;
11781173 const tunnels = new Map<string, SshTunnelEntry>();
1179 const pendingTunnelEntries = new Map<
1180 string,
1181 Deferred.Deferred<SshTunnelEntry, SshEnvironmentEffectError>
1182 >();
1174 const targetLocks = new Map<string, Semaphore.Semaphore>();
11831175 const authSecrets = new Map<string, string>();
11841176
1177 // Keep one lock per target so reconnect cannot reuse a server while stop is pending.
1178 const withTargetLock = Effect.fn("ssh/tunnel.withTargetLock")(function* <A, E, R>(
1179 key: string,
1180 effect: Effect.Effect<A, E, R>,
1181 ): Effect.fn.Return<A, E, R> {
1182 let lock = targetLocks.get(key);
1183 if (lock === undefined) {
1184 lock = Semaphore.makeUnsafe(1);
1185 targetLocks.set(key, lock);
1186 }
1187 return yield* lock.withPermits(1)(effect);
1188 });
1189
11851190 const closeTunnelEntry = Effect.fn("ssh/tunnel.closeTunnelEntry")(function* (
11861191 entry: SshTunnelEntry,
11871192 ) {
......@@ -1200,18 +1205,6 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
12001205 });
12011206 });
12021207
1203 const cancelPendingTunnelEntry = Effect.fn("ssh/tunnel.cancelPendingTunnelEntry")(function* (
1204 key: string,
1205 target: DesktopSshEnvironmentTarget,
1206 ) {
1207 const pending = pendingTunnelEntries.get(key);
1208 if (!pending) {
1209 return;
1210 }
1211 pendingTunnelEntries.delete(key);
1212 yield* Deferred.fail(pending, makeSshTunnelCancelledError(target)).pipe(Effect.ignore);
1213 });
1214
12151208 yield* Scope.addFinalizer(
12161209 managerScope,
12171210 Effect.sync(() => [...tunnels.values()]).pipe(
......@@ -1392,7 +1385,17 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
13921385 yield* Scope.addFinalizer(
13931386 entryScope,
13941387 Effect.gen(function* () {
1395 if (tunnels.get(tunnelEntry.key) !== tunnelEntry) {
1388 const stopRemote = tunnels.get(tunnelEntry.key) === tunnelEntry;
1389 if (stopRemote) {
1390 tunnels.delete(tunnelEntry.key);
1391 }
1392 yield* tunnelEntry.process
1393 .kill({
1394 killSignal: "SIGTERM",
1395 forceKillAfter: TUNNEL_SHUTDOWN_TIMEOUT_MS,
1396 })
1397 .pipe(Effect.ignore);
1398 if (!stopRemote) {
13961399 return;
13971400 }
13981401 yield* Effect.logDebug("ssh.environment.tunnel.finalizer.start", {
......@@ -1401,34 +1404,24 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
14011404 localPort: tunnelEntry.localPort,
14021405 remotePort: tunnelEntry.remotePort,
14031406 });
1404 tunnels.delete(tunnelEntry.key);
14051407 const authSecret = authSecrets.get(tunnelEntry.key) ?? null;
1406 yield* Effect.all(
1407 [
1408 tunnelEntry.process.kill({
1409 killSignal: "SIGTERM",
1410 forceKillAfter: TUNNEL_SHUTDOWN_TIMEOUT_MS,
1411 }),
1412 stopRemoteServer(
1413 tunnelEntry.target,
1414 authSecret === null
1415 ? {
1416 batchMode: "yes",
1417 interactiveAuth: false,
1418 }
1419 : {
1420 authSecret,
1421 batchMode: "no",
1422 interactiveAuth: true,
1423 },
1424 ).pipe(
1425 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawnerService),
1426 Effect.provideService(FileSystem.FileSystem, fileSystemService),
1427 Effect.provideService(Path.Path, pathService),
1428 ),
1429 ],
1430 { concurrency: "unbounded" },
1431 ).pipe(Effect.ignore);
1408 yield* stopRemoteServer(
1409 tunnelEntry.target,
1410 authSecret === null
1411 ? {
1412 batchMode: "yes",
1413 interactiveAuth: false,
1414 }
1415 : {
1416 authSecret,
1417 batchMode: "no",
1418 interactiveAuth: true,
1419 },
1420 ).pipe(
1421 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawnerService),
1422 Effect.provideService(FileSystem.FileSystem, fileSystemService),
1423 Effect.provideService(Path.Path, pathService),
1424 );
14321425 yield* Effect.logDebug("ssh.environment.tunnel.finalizer.succeeded", {
14331426 ...sshTargetLogFields(tunnelEntry.target),
14341427 key: tunnelEntry.key,
......@@ -1451,7 +1444,7 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
14511444 resolvedTarget: DesktopSshEnvironmentTarget,
14521445 runner?: RemoteT3RunnerOptions,
14531446 ): Effect.fn.Return<SshTunnelEntry, SshEnvironmentEffectError, SshEnvironmentEffectContext> {
1454 let entry = tunnels.get(key) ?? null;
1447 const entry = tunnels.get(key) ?? null;
14551448
14561449 if (entry !== null) {
14571450 yield* Effect.logDebug("ssh.environment.tunnel.existing.check", {
......@@ -1480,22 +1473,8 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
14801473 cause: readinessExit.cause,
14811474 });
14821475 yield* closeTunnelEntry(entry);
1483 yield* cancelPendingTunnelEntry(key, resolvedTarget);
1484 entry = null;
14851476 }
14861477
1487 const pending = pendingTunnelEntries.get(key);
1488 if (pending) {
1489 yield* Effect.logDebug("ssh.environment.tunnel.pending.await", {
1490 ...sshTargetLogFields(resolvedTarget),
1491 key,
1492 });
1493 return yield* Deferred.await(pending);
1494 }
1495
1496 const deferred = yield* Deferred.make<SshTunnelEntry, SshEnvironmentEffectError>();
1497 pendingTunnelEntries.set(key, deferred);
1498
14991478 return yield* createTunnelEntry({
15001479 key,
15011480 resolvedTarget,
......@@ -1508,13 +1487,6 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
15081487 cause,
15091488 }),
15101489 ),
1511 Effect.onExit((exit) =>
1512 Effect.sync(() => {
1513 if (pendingTunnelEntries.get(key) === deferred) {
1514 pendingTunnelEntries.delete(key);
1515 }
1516 }).pipe(Effect.andThen(Deferred.done(deferred, exit))),
1517 ),
15181490 );
15191491 });
15201492
......@@ -1553,33 +1525,39 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
15531525 ...sshRunnerLogFields(runner),
15541526 key,
15551527 });
1556 const entry = yield* ensureTunnelEntry(key, resolvedTarget, runner);
1528 return yield* withTargetLock(
1529 key,
1530 Effect.gen(function* () {
1531 const entry = yield* ensureTunnelEntry(key, resolvedTarget, runner);
1532
1533 const pairingResult = requestOptions?.issuePairingToken
1534 ? yield* runWithSshAuth({
1535 key,
1536 target: entry.target,
1537 operation: (authOptions) =>
1538 issueRemotePairingToken(entry.target, authOptions, runner),
1539 })
1540 : null;
1541 const pairingToken = pairingResult?.credential ?? null;
15571542
1558 const pairingResult = requestOptions?.issuePairingToken
1559 ? yield* runWithSshAuth({
1543 yield* Effect.logInfo("ssh.environment.ensure.succeeded", {
1544 ...sshTargetLogFields(entry.target),
15601545 key,
1546 localPort: entry.localPort,
1547 remotePort: entry.remotePort,
1548 remoteServerKind: entry.remoteServerKind,
1549 issuedPairingToken: pairingToken !== null,
1550 });
1551 return {
15611552 target: entry.target,
1562 operation: (authOptions) => issueRemotePairingToken(entry.target, authOptions, runner),
1563 })
1564 : null;
1565 const pairingToken = pairingResult?.credential ?? null;
1566
1567 yield* Effect.logInfo("ssh.environment.ensure.succeeded", {
1568 ...sshTargetLogFields(entry.target),
1569 key,
1570 localPort: entry.localPort,
1571 remotePort: entry.remotePort,
1572 remoteServerKind: entry.remoteServerKind,
1573 issuedPairingToken: pairingToken !== null,
1574 });
1575 return {
1576 target: entry.target,
1577 httpBaseUrl: entry.httpBaseUrl,
1578 wsBaseUrl: entry.wsBaseUrl,
1579 pairingToken,
1580 remotePort: entry.remotePort,
1581 ...(entry.remoteServerKind ? { remoteServerKind: entry.remoteServerKind } : {}),
1582 };
1553 httpBaseUrl: entry.httpBaseUrl,
1554 wsBaseUrl: entry.wsBaseUrl,
1555 pairingToken,
1556 remotePort: entry.remotePort,
1557 ...(entry.remoteServerKind ? { remoteServerKind: entry.remoteServerKind } : {}),
1558 };
1559 }),
1560 );
15831561 });
15841562
15851563 const disconnectEnvironment = Effect.fn("ssh/tunnel.disconnectEnvironment")(function* (
......@@ -1593,28 +1571,33 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
15931571 ...(target.port !== null ? { port: target.port } : {}),
15941572 };
15951573 const key = targetConnectionKey(resolvedTarget);
1596 const entry = tunnels.get(key) ?? null;
1597 yield* Effect.logDebug("ssh.environment.disconnect.targetResolved", {
1598 ...sshTargetLogFields(resolvedTarget),
1599 key,
1600 hasTunnel: entry !== null,
1601 hasPendingTunnel: pendingTunnelEntries.has(key),
1602 });
1603 if (entry !== null) {
1604 yield* closeTunnelEntry(entry);
1605 }
1606 yield* cancelPendingTunnelEntry(key, resolvedTarget);
1607 if (entry === null) {
1608 yield* runWithSshAuth({
1609 key,
1610 target: resolvedTarget,
1611 operation: (authOptions) => stopRemoteServer(resolvedTarget, authOptions),
1612 });
1613 }
1614 yield* Effect.logInfo("ssh.environment.disconnect.succeeded", {
1615 ...sshTargetLogFields(resolvedTarget),
1574 yield* withTargetLock(
16161575 key,
1617 });
1576 Effect.gen(function* () {
1577 const entry = tunnels.get(key) ?? null;
1578 yield* Effect.logDebug("ssh.environment.disconnect.targetResolved", {
1579 ...sshTargetLogFields(resolvedTarget),
1580 key,
1581 hasTunnel: entry !== null,
1582 });
1583 if (entry !== null) {
1584 // Explicit disconnect owns the remote stop so its failure reaches the caller.
1585 yield* Effect.gen(function* () {
1586 tunnels.delete(key);
1587 yield* closeTunnelEntry(entry);
1588 }).pipe(Effect.uninterruptible);
1589 }
1590 yield* runWithSshAuth({
1591 key,
1592 target: resolvedTarget,
1593 operation: (authOptions) => stopRemoteServer(resolvedTarget, authOptions),
1594 });
1595 yield* Effect.logInfo("ssh.environment.disconnect.succeeded", {
1596 ...sshTargetLogFields(resolvedTarget),
1597 key,
1598 });
1599 }),
1600 );
16181601 });
16191602
16201603 return SshEnvironmentManager.of({ ensureEnvironment, disconnectEnvironment });