🤖 feat: SSH connection pool with backoff and singleflighting (#922)

Prevents thundering herd issues with SSH connections by adding health
tracking, exponential backoff, and singleflighting to the connection
pool.

## Changes

- **SSHConnectionPool class** with:
  - Health tracking (healthy/unhealthy/unknown states)
  - Exponential backoff: 1s → 5s → 10s → 20s → 40s → 60s (cap)
  - Singleflighting: concurrent probes to same host share one attempt
  - Fast-path for known-healthy connections (no re-probe)

- **Integration points**:
- `SSHRuntime.exec()` and `execSSHCommand()` call `acquireConnection()`
- `PTYService` calls `acquireConnection()` before spawning SSH terminals

## Flow

```
acquireConnection() → in backoff? → throw immediately
                   → known healthy? → return immediately  
                   → inflight probe? → wait on existing promise
                   → start probe → success? → mark healthy, return
                                 → failure? → mark failed + backoff, throw
```

_Generated with `mux`_

Author: deansheather

src/node/runtime/SSHRuntime.ts

@@ -24,7 +24,7 @@ import { expandTildeForSSH, cdCommandForSSH } from "./tildeExpansion";
 import { getProjectName, execBuffered } from "@/node/utils/runtime/helpers";
 import { getErrorMessage } from "@/common/utils/errors";
 import { execAsync, DisposableProcess } from "@/node/utils/disposableExec";
-import { getControlPath } from "./sshConnectionPool";
+import { getControlPath, sshConnectionPool, type SSHRuntimeConfig } from "./sshConnectionPool";
 import { getBashPath } from "@/node/utils/main/bashPath";
 
 /**
@@ -41,21 +41,8 @@ const shescape = {
   },
 };
 
-/**
- * SSH Runtime Configuration
- */
-export interface SSHRuntimeConfig {
-  /** SSH host (can be hostname, user@host, or SSH config alias) */
-  host: string;
-  /** Working directory on remote host */
-  srcBaseDir: string;
-  /** Directory on remote for background process output (default: /tmp/mux-bashes) */
-  bgOutputDir?: string;
-  /** Optional: Path to SSH private key (if not using ~/.ssh/config or ssh-agent) */
-  identityFile?: string;
-  /** Optional: SSH port (default: 22) */
-  port?: number;
-}
+// Re-export SSHRuntimeConfig from connection pool (defined there to avoid circular deps)
+export type { SSHRuntimeConfig } from "./sshConnectionPool";
 
 /**
  * SSH runtime implementation that executes commands and file operations
@@ -127,7 +114,6 @@ export class SSHRuntime implements Runtime {
   /**
    * Execute command over SSH with streaming I/O
    */
-  // eslint-disable-next-line @typescript-eslint/require-await
   async exec(command: string, options: ExecOptions): Promise<ExecStream> {
     const startTime = performance.now();
 
@@ -136,6 +122,10 @@ export class SSHRuntime implements Runtime {
       throw new RuntimeErrorClass("Operation aborted before execution", "exec");
     }
 
+    // Ensure connection is healthy before executing
+    // This provides backoff protection and singleflighting for concurrent requests
+    await sshConnectionPool.acquireConnection(this.config);
+
     // Build command parts
     const parts: string[] = [];
 
@@ -233,11 +223,22 @@ export class SSHRuntime implements Runtime {
           resolve(EXIT_CODE_TIMEOUT);
           return;
         }
-        resolve(code ?? (signal ? -1 : 0));
+
+        const exitCode = code ?? (signal ? -1 : 0);
+
+        // SSH exit code 255 indicates connection failure - report to pool for backoff
+        // This prevents thundering herd when a previously healthy host goes down
+        if (exitCode === 255) {
+          sshConnectionPool.reportFailure(this.config, "SSH connection failed (exit code 255)");
+        }
+
+        resolve(exitCode);
         // Cleanup runs automatically via DisposableProcess
       });
 
       sshProcess.on("error", (err) => {
+        // Spawn errors are connection-level failures
+        sshConnectionPool.reportFailure(this.config, `SSH spawn error: ${err.message}`);
         reject(new RuntimeErrorClass(`Failed to execute SSH command: ${err.message}`, "exec", err));
       });
     });
@@ -427,6 +428,9 @@ export class SSHRuntime implements Runtime {
    * @private
    */
   private async execSSHCommand(command: string, timeoutMs: number): Promise<string> {
+    // Ensure connection is healthy before executing
+    await sshConnectionPool.acquireConnection(this.config, timeoutMs);
+
     const sshArgs = this.buildSSHArgs();
     sshArgs.push(this.config.host, command);
 
@@ -461,6 +465,10 @@ export class SSHRuntime implements Runtime {
         if (timedOut) return; // Already rejected
 
         if (code !== 0) {
+          // SSH exit code 255 indicates connection failure - report to pool for backoff
+          if (code === 255) {
+            sshConnectionPool.reportFailure(this.config, "SSH connection failed (exit code 255)");
+          }
           reject(new RuntimeErrorClass(`SSH command failed: ${stderr.trim()}`, "network"));
           return;
         }
@@ -473,6 +481,8 @@ export class SSHRuntime implements Runtime {
         clearTimeout(timer);
         if (timedOut) return; // Already rejected
 
+        // Spawn errors are connection-level failures
+        sshConnectionPool.reportFailure(this.config, `SSH spawn error: ${getErrorMessage(err)}`);
         reject(
           new RuntimeErrorClass(
             `Cannot execute SSH command: ${getErrorMessage(err)}`,

src/node/runtime/sshConnectionPool.test.ts

@@ -1,7 +1,6 @@
 import * as os from "os";
 import * as path from "path";
-import { getControlPath } from "./sshConnectionPool";
-import type { SSHRuntimeConfig } from "./SSHRuntime";
+import { getControlPath, SSHConnectionPool, type SSHRuntimeConfig } from "./sshConnectionPool";
 
 describe("sshConnectionPool", () => {
   describe("getControlPath", () => {
@@ -59,7 +58,9 @@ describe("sshConnectionPool", () => {
       expect(getControlPath(config1)).not.toBe(getControlPath(config2));
     });
 
-    test("different srcBaseDirs produce different controlPaths", () => {
+    test("different srcBaseDirs produce same controlPaths (connection shared)", () => {
+      // srcBaseDir is intentionally excluded from connection key -
+      // workspaces on the same host share health tracking and multiplexing
       const config1: SSHRuntimeConfig = {
         host: "test.com",
         srcBaseDir: "/work1",
@@ -69,7 +70,7 @@ describe("sshConnectionPool", () => {
         srcBaseDir: "/work2",
       };
 
-      expect(getControlPath(config1)).not.toBe(getControlPath(config2));
+      expect(getControlPath(config1)).toBe(getControlPath(config2));
     });
 
     test("controlPath is in tmpdir with expected format", () => {
@@ -134,3 +135,151 @@ describe("username isolation", () => {
     expect(controlPath).toMatch(/mux-ssh-[a-f0-9]{12}$/);
   });
 });
+
+describe("SSHConnectionPool", () => {
+  describe("health tracking", () => {
+    test("getConnectionHealth returns undefined for unknown connection", () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "unknown.example.com",
+        srcBaseDir: "/work",
+      };
+
+      expect(pool.getConnectionHealth(config)).toBeUndefined();
+    });
+
+    test("markHealthy sets connection to healthy state", () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "test.example.com",
+        srcBaseDir: "/work",
+      };
+
+      pool.markHealthy(config);
+      const health = pool.getConnectionHealth(config);
+
+      expect(health).toBeDefined();
+      expect(health!.status).toBe("healthy");
+      expect(health!.consecutiveFailures).toBe(0);
+      expect(health!.lastSuccess).toBeInstanceOf(Date);
+    });
+
+    test("reportFailure puts connection into backoff", () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "test.example.com",
+        srcBaseDir: "/work",
+      };
+
+      // Mark healthy first
+      pool.markHealthy(config);
+      expect(pool.getConnectionHealth(config)?.status).toBe("healthy");
+
+      // Report a failure
+      pool.reportFailure(config, "Connection refused");
+      const health = pool.getConnectionHealth(config);
+
+      expect(health?.status).toBe("unhealthy");
+      expect(health?.consecutiveFailures).toBe(1);
+      expect(health?.lastError).toBe("Connection refused");
+      expect(health?.backoffUntil).toBeDefined();
+    });
+
+    test("resetBackoff clears backoff state after failed probe", async () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "nonexistent.invalid.host.test",
+        srcBaseDir: "/work",
+      };
+
+      // Trigger a failure via acquireConnection (will fail to connect)
+      await expect(pool.acquireConnection(config, 1000)).rejects.toThrow();
+
+      // Verify we're now in backoff
+      const healthBefore = pool.getConnectionHealth(config);
+      expect(healthBefore?.status).toBe("unhealthy");
+      expect(healthBefore?.backoffUntil).toBeDefined();
+
+      // Reset backoff
+      pool.resetBackoff(config);
+      const healthAfter = pool.getConnectionHealth(config);
+
+      expect(healthAfter).toBeDefined();
+      expect(healthAfter!.status).toBe("unknown");
+      expect(healthAfter!.consecutiveFailures).toBe(0);
+      expect(healthAfter!.backoffUntil).toBeUndefined();
+    });
+  });
+
+  describe("acquireConnection", () => {
+    test("returns immediately for known healthy connection", async () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "test.example.com",
+        srcBaseDir: "/work",
+      };
+
+      // Mark as healthy first
+      pool.markHealthy(config);
+
+      // Should return immediately without probing
+      const start = Date.now();
+      await pool.acquireConnection(config);
+      const elapsed = Date.now() - start;
+
+      // Should be nearly instant (< 50ms)
+      expect(elapsed).toBeLessThan(50);
+    });
+
+    test("throws immediately when in backoff", async () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "nonexistent.invalid.host.test",
+        srcBaseDir: "/work",
+      };
+
+      // Trigger a failure to put connection in backoff
+      await expect(pool.acquireConnection(config, 1000)).rejects.toThrow();
+
+      // Second call should throw immediately with backoff message
+      await expect(pool.acquireConnection(config)).rejects.toThrow(/in backoff/);
+    });
+
+    test("getControlPath returns deterministic path", () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "test.example.com",
+        srcBaseDir: "/work",
+      };
+
+      const path1 = pool.getControlPath(config);
+      const path2 = pool.getControlPath(config);
+
+      expect(path1).toBe(path2);
+      expect(path1).toBe(getControlPath(config));
+    });
+  });
+
+  describe("singleflighting", () => {
+    test("concurrent acquireConnection calls share same probe", async () => {
+      const pool = new SSHConnectionPool();
+      const config: SSHRuntimeConfig = {
+        host: "nonexistent.invalid.host.test",
+        srcBaseDir: "/work",
+      };
+
+      // All concurrent calls should share the same probe and get same result
+      const results = await Promise.allSettled([
+        pool.acquireConnection(config, 1000),
+        pool.acquireConnection(config, 1000),
+        pool.acquireConnection(config, 1000),
+      ]);
+
+      // All should be rejected (connection fails)
+      expect(results.every((r) => r.status === "rejected")).toBe(true);
+
+      // Only 1 failure should be recorded (not 3) - proves singleflighting worked
+      expect(pool.getConnectionHealth(config)?.consecutiveFailures).toBe(1);
+    });
+  });
+});