Jens/ForgeFlow
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 24 Wiki Activity

perf: reuse SSH connections per server, with a retry rule that never repeats work

Browse Source 
Every ssh.exec opened its own connection: a TCP handshake, a key exchange and an
authentication round trip per command. A key rotation paid for that eight times,
a deployment six, and refreshing M profile states M times.

Connections are now kept per server. The three risks that made this worth doing
carefully are handled explicitly:

- Staleness. A pooled connection can be dead exactly when it matters. Liveness is
  tracked through error, close and end, and a lease that finds a dead entry opens
  a new one. The remaining race, where the connection dies between the check and
  the command, is caught by the retry rule below.
- Retrying. Only a failure that proves the command never reached the server is
  retried, and only once, and only on a connection that was already established
  before this call. execClient marks exactly that case, when the channel fails to
  open. A command that opened a stream is never repeated, because the server may
  already be acting on it - repeating a deployment is not this layer's decision.
  Two tests hold that line: widening the rule to any failure fails both.
- Lifetime. Idle connections close after a minute, the pool is reference counted
  so a shared connection survives until its last user is done, closeAll runs
  during quit, and every pooled client keeps a standing error listener so an
  error while idle cannot reach the uncaughtException handler.

A trust-on-first-use connection is never pooled: it was established without
verifying the fingerprint, so it must not serve a later verified call. A change
to host, port, user, auth type, key path or trusted fingerprint invalidates the
pooled connection.

ssh-service coverage rises from 61% to 90% of lines and 97% of functions.

Also in this commit, the smaller items from the same review:

- Diagnostics batched records that queue up while a write is in flight into one
  append, and chmod runs once per file instead of once per record. At the debug
  level every IPC call writes a line, which is exactly when troubleshooting.
- The set that suppresses duplicate deployment notifications is trimmed instead
  of growing for the lifetime of the process.
- The updater kept the same once('error') pattern on its spawned helper that
  took the app down through the SSH client.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
NuklearRabbit
2026-08-23 15:06:38 +02:00
co-authored by Claude Opus 5
parent beeafdcba7
commit cb9bdcd713
8 changed files with 470 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SOURCE_MANIFEST.txt
+7 -6
View File
@@ -104,7 +104,7 @@ c230b931abf2293d2d44b7a69b94c35f1142c093cc46b88739a0de5cbd6d1896 1532
106538d4a14a5a7b13419f9520c582b19809e8fafe2cb8c7dce2bc3e600dd10a 397 examples/server/nginx-forgeflow-status.conf
2dff25fb39ce8fc7844026a50524b23f241bec5b614eb05371c7f908a080f69a 398 examples/server/status-example.json
4a561ead5ba7cdfaf4efce91842a4308c5f2a77980205879d83835efb8a579db 1067 LICENSE
d7cae22a646c8f06f339db799f2b9921ba1799cdc44e5d05578c04aaace103ab 14357 main.cjs
e2daa28bbc01c68c3702add6ea8259dff5920b22f6fdc3c9193ed78a153f2e9e 14708 main.cjs
91a984a89dd57a084b9a2331763cacdb061582fb590f13df379d92c1a77a2ee1 352 OVERLAY-INSTRUCTIONS.md
198a8a51705fcea2ddf716558a49e06df5a8181c5476709c474d183b64890ca0 179808 package-lock.json
fb12ca4da1105da46b4aacacb67260357c29aa362385c9cfaaeadeffd1bf82b3 6142 package.json
@@ -114,8 +114,8 @@ abe5dd6fd68f2970cd19ef134094907c67219061d8fe9a1a08324c78de4ad437 484
f018383f755352ca448e2ebb1e19b1dba412a3eb793d61e64b02953e300754fd 10538 Publish-ForgeFlow-Release.ps1
688fff7d2c989adb97ebb7fae38962656b70304a0aa5d27433c56adf7f136de0 4196 Publish-Missing-Binary-Release.ps1
42e2f375c2fa9bea138428b85eb2bb93fe8bf6e4a5fc0cd728f2c6713a019df7 10315 README.md
64d4c37123a87704e584cd40b3cdadbbece768ca36cc6f10c6666f266be056c7 14947 reports/architecture-audit.json
1ed7d86361db5adbb71f1dd042f66e62e6a2d0b774db59bcb5f215b5996c17b5 1343 reports/architecture-audit.md
bfddedaf5e21a4948a4ba3d0ca89aee968028990f0079748963d4088f22d4537 14962 reports/architecture-audit.json
68c2e5d245f93bb6c8e53bed105ed827601443827972b157a60b858aa440afaf 1343 reports/architecture-audit.md
509c7bcff5280349bd9f45ed6151f70372bad7010a9ea582c13e2ccab91fe0cd 6272 scripts/acceptance.mjs
d0745072321aca2c80f44460974a7926715a9f429164aaf7660dced40b52c736 4790 scripts/apply-binary-update.ps1
404863bcbe7292355662e3a326455df864d7279badc29f90866a3b837420df54 10745 scripts/apply-source-update.ps1
@@ -138,7 +138,7 @@ dd613d04b366f2cd071a1685a414016a5fb008082ed1b4cb8b24b79c100f640a 2412
86e9fc2eda66b4b563f6c4bbb87d3e8514340d484fb503b73137e63b6b05c3c9 14597 src/main/deploy-key-lifecycle-service.cjs
7cbfe51973d6607203cb197652ed7f296a3f6b6b644df876957117866a47d802 2159 src/main/deployment-identity.cjs
ce30ddac403d1adf21176e5df21b0cc3db435305d2628f51f1486eacf20df6f2 23708 src/main/deployment-service.cjs
c157640e76d558906a9aa9881eda811196623ef1c65fa3467f32f0f84b0ddd0c 15095 src/main/diagnostics-service.cjs
165a37f8fbc82fb17e66c177f32aef70ac1ea293b00b82fc9fe301fe7e8b7080 15943 src/main/diagnostics-service.cjs
a2ef47d5330095b92c2bd22fcc39962091881f9cb60d02e261eb1dd1bd693170 1974 src/main/external-tools-service.cjs
665859c48be68e5f256b9fac9a4722390cf1154705c099778f36e0e0a6c07e27 35214 src/main/git-service.cjs
e28fc1ca2fd4c0116148f5005d793feddf04c36ef711d2d348560394d209a613 7253 src/main/git-validator-policy.cjs
@@ -159,7 +159,7 @@ e64f7257d478955c675a133b3735b6afe138a69d2ad090898061e56f557c43e5 9926
c629efa832a806409954ab5a4986bf88606ce5fa5278bb1885a76222d1cb6faf 6441 src/main/repository-monitor.cjs
6393583911263575c6e2a19d9baab6e638cce90252c386b0a5144f2fb6f81f15 12154 src/main/repository-service.cjs
52b6d88ed1f5c904a13cdde92e5f96d1e2b5971ceef49862152197353cdc6490 27928 src/main/server-inventory.cjs
c7921a2380e305abdabb94dc0d04c6060eaa09bbe01f18994786f20bfa02d0c3 15107 src/main/ssh-service.cjs
792c5cf1b884e71a601b80ee373ff81178c34a910da5a7d64d3040f266bb0cee 20398 src/main/ssh-service.cjs
19538a3c40ea3489bbaee9a23af36a5e99962af6bb3d04259f05ece6588cbeb2 25901 src/main/unraid-access-methods.cjs
5621e35323e4f81fb14a05670f81579ec1e66bea3a55fa6457ece0f807421424 9801 src/main/unraid-deploy-key-host.cjs
6d9910dace52625f88e066a8485af2663c3735ff15e9ce9031441ce742710a21 30793 src/main/unraid-deployment-methods.cjs
@@ -168,7 +168,7 @@ a2fa253bb2664ad7e430d7a4cd4c4434b89e6c54c19c7b9e4f53fe08485339d3 42652
2c0cf07921ca7ee5a9085ced44498c2e6798e5cc1e8a5ecf704c3cecabe39a25 27607 src/main/unraid-preflight-methods.cjs
d45220176aed72d692f9ae5534f9d40bcc359a2d08e025e74a3b3b505b8b9ed4 16559 src/main/unraid-runtime-methods.cjs
4c5cf01922e1feb36a31b50af22e973d8aee3fecccd406e449690604111898ac 11608 src/main/unraid-state-methods.cjs
a387ac9735624ef89d964ea59e035dae256b9915ba2cee22ae5cbc2894f5808a 22000 src/main/update-service.cjs
d968d64d1e65f4a0fbb7a19f16775469d8e2e35b5ae249813c0b635b27bef8fe 22311 src/main/update-service.cjs
b5c304531bec358d059189a27cd9db8fa20cefb7f817e5eb0287001f7353f6a7 985 src/renderer/actions/command.js
d0bf607dd1de9d55f2947d0adf0997cd3ca5c269d10a5362cc1d8bc4d1a2a8ae 6706 src/renderer/actions/deployment-operation.js
33e65457e44708cecbe859b8c8a0c51b4ab33219c3b7a9f9dad884089f07d367 17904 src/renderer/actions/deployment-profile.js
@@ -237,6 +237,7 @@ d49c772e3c7ddaa12dc5a1d4fc4cb474a4d99ae06fa5dab5a6cf1c44acb9ed6f 3463
bab853feb0e22aa25af17989baaa632c01efa636533ea67407fecfdd973c7024 627 tests/semver.test.mjs
12cb3b240bdd0922566323c0014838ca067ad10d9d4009943165ae2c4e93bc6f 11786 tests/server-inventory-branches.test.mjs
020eccfa9c4aef7a4ac4736d9af90518fcb6d1ad75aedcfaa1c92832a9e3d6d8 4609 tests/shell-verification.test.mjs
a39d30f47813dfb98c998811f3d76ebbb1544ecfd017a165d44f9afb80d7daf9 9090 tests/ssh-connection-pool.test.mjs
7ee9166327ed227d2b7c6929692dea5c5d7a41c3e566596fa92d9ec4f42e8677 4085 tests/ssh-connection.test.mjs
0d1bc4d623ce299337736c577ec61c8ffd6974ebe20335b72838d10eae35ecb1 7993 tests/ssh-service.test.mjs
8a6a8477eb94b85ccef18cddd2640afb0d1eafa679c96bc7de20428d5d69e1be 1794 tests/tool-invocation.test.mjs
main.cjs
+8
View File
@@ -40,6 +40,7 @@ const {
let mainWindow;
let repositoryMonitor;
let sshService;
let operationTimer;
let diagnostics;
let configStore;
@@ -252,6 +253,7 @@ app
const repositories = new RepositoryService(store, git, gitea, diagnostics);
const deployments = new DeploymentService(store, gitea, git, diagnostics);
const ssh = new SshService({ store, diagnostics });
sshService = ssh;
const auditedOperationStates = new Set();
const reportOperationChange = (payload) => {
broadcast("operations:changed", payload);
@@ -262,6 +264,11 @@ app
) {
const key = `${operation.id}:${operation.status}`;
if (!auditedOperationStates.has(key)) {
// One entry per completed deployment, so the set is trimmed rather
// than kept for the lifetime of the process.
if (auditedOperationStates.size >= 500) {
auditedOperationStates.delete(auditedOperationStates.values().next().value);
}
auditedOperationStates.add(key);
notify(
`Deployment ${operation.status}`,
@@ -450,6 +457,7 @@ app.on("before-quit", (event) => {
event.preventDefault();
quitCleanupStarted = true;
repositoryMonitor?.stop();
sshService?.closeAll();
if (operationTimer) clearTimeout(operationTimer);
Promise.resolve()
.then(() => diagnostics?.info("app.quitting", {}))
reports/architecture-audit.json
+39 -38
View File
@@ -1,5 +1,5 @@
{
"generatedAt": "2026-08-23T12:51:31.496Z",
"generatedAt": "2026-08-23T13:06:16.630Z",
"thresholds": {
"preferredMaximumLines": 750,
"justificationRequiredLines": 1000
@@ -206,6 +206,23 @@
],
"hotspotScore": 91
},
{
"file": "main.cjs",
"lines": 471,
"branches": 39,
"functions": 57,
"ipcHandlers": 0,
"responsibilities": [
"inventory",
"deployment",
"git",
"ipc",
"renderer",
"security",
"updates"
],
"hotspotScore": 89
},
{
"file": "src/renderer/actions/setup-and-settings.js",
"lines": 427,
@@ -236,21 +253,18 @@
"hotspotScore": 87
},
{
"file": "main.cjs",
"lines": 463,
"branches": 37,
"functions": 56,
"file": "src/main/ssh-service.cjs",
"lines": 467,
"branches": 66,
"functions": 92,
"ipcHandlers": 0,
"responsibilities": [
"inventory",
"deployment",
"git",
"ipc",
"renderer",
"security",
"updates"
],
"hotspotScore": 87
"hotspotScore": 86
},
{
"file": "src/renderer/actions/deployment-profile.js",
@@ -343,6 +357,21 @@
],
"hotspotScore": 73
},
{
"file": "src/main/diagnostics-service.cjs",
"lines": 373,
"branches": 39,
"functions": 51,
"ipcHandlers": 0,
"responsibilities": [
"deployment",
"git",
"ipc",
"security",
"updates"
],
"hotspotScore": 69
},
{
"file": "src/renderer/events.js",
"lines": 181,
@@ -372,20 +401,6 @@
],
"hotspotScore": 63
},
{
"file": "src/main/ssh-service.cjs",
"lines": 336,
"branches": 43,
"functions": 69,
"ipcHandlers": 0,
"responsibilities": [
"deployment",
"git",
"security",
"updates"
],
"hotspotScore": 63
},
{
"file": "src/main/deploy-key-lifecycle-service.cjs",
"lines": 192,
@@ -415,20 +430,6 @@
],
"hotspotScore": 55
},
{
"file": "src/main/diagnostics-service.cjs",
"lines": 357,
"branches": 34,
"functions": 50,
"ipcHandlers": 0,
"responsibilities": [
"deployment",
"git",
"security",
"updates"
],
"hotspotScore": 54
},
{
"file": "src/renderer/actions/inventory.js",
"lines": 186,
@@ -668,7 +669,7 @@
},
{
"file": "src/main/unraid-deploy-key-host.cjs",
"lines": 77,
"lines": 80,
"branches": 7,
"functions": 22,
"ipcHandlers": 0,
reports/architecture-audit.md
+1 -1
View File
@@ -1,6 +1,6 @@
# ForgeFlow architecture audit
Generated 2026-08-23T12:51:31.496Z. Complexity is a deterministic decision-point count used for hotspot ranking, not a claim of exact McCabe complexity.
Generated 2026-08-23T13:06:16.630Z. Complexity is a deterministic decision-point count used for hotspot ranking, not a claim of exact McCabe complexity.
## Files above 750 lines
src/main/diagnostics-service.cjs
+21 -5
View File
@@ -46,6 +46,9 @@ class DiagnosticsService {
this.preferencesProvider = preferencesProvider;
this.sessionId = crypto.randomUUID();
this.writeChain = Promise.resolve();
this.pendingLines = [];
this.pendingFlush = null;
this.securedFiles = new Set();
this.initialized = false;
this.lastWriteError = null;
this.lastBundlePath = null;
@@ -115,13 +118,25 @@ class DiagnosticsService {
sessionId: this.sessionId,
details
});
const line = `${JSON.stringify(record)}\n`;
this.writeChain = this.writeChain.then(async () => {
this.pendingLines.push(`${JSON.stringify(record)}\n`);
// At the debug level every IPC call and every Gitea request writes a line.
// Records that queue up while a write is in flight are appended together, so
// a burst costs one open/write/close instead of one per record.
if (this.pendingFlush) return this.pendingFlush;
this.pendingFlush = this.writeChain.then(async () => {
this.pendingFlush = null;
const lines = this.pendingLines.splice(0).join('');
if (!lines) return true;
try {
if (!this.initialized) await fs.mkdir(this.logDirectory, { recursive: true, mode: 0o700 });
const target = await this.rotateIfNeeded(this.filePathForToday());
await fs.appendFile(target, line, { encoding: 'utf8', mode: 0o600 });
try { await fs.chmod(target, 0o600); } catch {}
await fs.appendFile(target, lines, { encoding: 'utf8', mode: 0o600 });
// The mode above only applies when appendFile creates the file, so the
// explicit chmod is needed once per file rather than once per record.
if (!this.securedFiles.has(target)) {
try { await fs.chmod(target, 0o600); } catch { /* best effort */ }
this.securedFiles.add(target);
}
this.lastWriteError = null;
return true;
} catch (error) {
@@ -129,7 +144,8 @@ class DiagnosticsService {
return false;
}
});
return this.writeChain;
this.writeChain = this.pendingFlush.catch(() => {});
return this.pendingFlush;
}
debug(event, details) { return this.log('debug', event, details); }
src/main/ssh-service.cjs
+133 -2
View File
@@ -53,9 +53,28 @@ function parseCapabilityOutput(output) {
}
class SshService {
constructor({ store, diagnostics }) {
constructor({ store, diagnostics, idleConnectionMs = 60_000 }) {
this.store = store;
this.diagnostics = diagnostics;
// Every command used to pay for a TCP handshake, a key exchange and an
// authentication round trip. Sessions are kept per server for a short while
// so a sequence of commands shares one connection.
this.sessions = new Map();
this.idleConnectionMs = idleConnectionMs;
}
// A connection is only reusable for a server whose identity and credentials
// are unchanged. Anything in this key changing means a new connection.
sessionKey(server) {
return JSON.stringify([
server.id,
server.host,
server.port || 22,
server.username,
server.authType,
server.privateKeyPath || '',
server.hostFingerprint || '',
]);
}
async validateServerConfiguration(server, secrets = {}) {
@@ -116,6 +135,114 @@ class SshService {
async withClient(serverId, action, options = {}) {
const server = this.store.getServer(serverId);
if (!server) throw new Error('The configured SSH server no longer exists.');
// A trust-on-first-use connection is established without checking the
// fingerprint, so it must never serve a later verified call.
if (options.trustOnFirstUse) return this.withDedicatedClient(server, action, options);
return this.withPooledClient(server, action, options);
}
// Retrying is only safe while the command has not reached the server. Once a
// stream is open the remote side may already be deploying, and repeating that
// is not something this layer is allowed to decide.
isPreCommandFailure(error) {
return error?.beforeCommand === true;
}
async withPooledClient(server, action, options) {
const key = this.sessionKey(server);
for (let attempt = 0; ; attempt += 1) {
const session = await this.leaseSession(server, key, options);
try {
const result = await action(session.client, server, session.fingerprint);
this.releaseSession(session);
return result;
} catch (error) {
const staleConnection = session.reused && attempt === 0 && this.isPreCommandFailure(error);
this.discardSession(session);
if (!staleConnection) throw error;
await this.diagnostics?.debug('ssh.session.stale-retry', { serverId: server.id, host: server.host, message: error.message });
}
}
}
createSession(server, key, options) {
const entry = { key, client: null, fingerprint: null, leases: 0, dead: false, established: false, idleTimer: null, opening: null };
entry.opening = this
.withDedicatedClient(server, async (client, _server, fingerprint) => ({ client, fingerprint }), options, { keepOpen: true })
.then((opened) => {
entry.client = opened.client;
entry.fingerprint = opened.fingerprint;
entry.established = true;
// Without a standing listener an error on an idle connection is
// unhandled, which terminates the main process.
opened.client.on('error', () => this.markSessionDead(entry));
opened.client.on('close', () => this.markSessionDead(entry));
opened.client.on('end', () => this.markSessionDead(entry));
});
this.sessions.set(key, entry);
return entry;
}
async leaseSession(server, key, options) {
const pooled = this.sessions.get(key);
// Only a connection that was already up before this call may be retried on
// failure. Callers that arrive while one is still being opened share both
// the connection and its outcome.
const reused = Boolean(pooled && !pooled.dead && pooled.established);
const entry = pooled && !pooled.dead ? pooled : this.createSession(server, key, options);
entry.leases += 1;
if (entry.idleTimer) { clearTimeout(entry.idleTimer); entry.idleTimer = null; }
try {
await entry.opening;
} catch (error) {
entry.leases -= 1;
this.markSessionDead(entry);
throw error;
}
return { client: entry.client, fingerprint: entry.fingerprint, reused, entry };
}
markSessionDead(entry) {
entry.dead = true;
if (this.sessions.get(entry.key) === entry) this.sessions.delete(entry.key);
if (entry.idleTimer) { clearTimeout(entry.idleTimer); entry.idleTimer = null; }
if (entry.leases <= 0) this.endSession(entry);
}
endSession(entry) {
if (!entry.client) return;
try { entry.client.end(); } catch { /* already closed */ }
}
releaseSession(session) {
const entry = session.entry;
entry.leases -= 1;
if (entry.dead) { if (entry.leases <= 0) this.endSession(entry); return; }
if (entry.leases > 0) return;
entry.idleTimer = setTimeout(() => {
entry.idleTimer = null;
this.markSessionDead(entry);
}, this.idleConnectionMs);
entry.idleTimer.unref?.();
}
discardSession(session) {
const entry = session.entry;
entry.leases -= 1;
this.markSessionDead(entry);
}
// Closes every pooled connection. The application calls this while quitting so
// no socket outlives the process.
closeAll() {
for (const entry of [...this.sessions.values()]) {
entry.leases = 0;
this.markSessionDead(entry);
}
}
async withDedicatedClient(server, action, options = {}, { keepOpen = false } = {}) {
const serverId = server.id;
const Client = loadSshClient();
const connection = await this.connectionOptions(server, options);
const client = new Client();
@@ -125,7 +252,8 @@ class SshService {
const finish = (callback, value) => {
if (settled) return;
settled = true;
try { client.end(); } catch {}
// A session that stays in the pool is closed by the pool, not here.
if (!(keepOpen && callback === resolve)) { try { client.end(); } catch { /* already closed */ } }
callback(value);
};
client.once('ready', async () => {
@@ -167,6 +295,9 @@ class SshService {
if (error) {
clearTimeout(timer);
completed = true;
// The channel never opened, so the command did not reach the server.
// This is the only failure the pool is allowed to retry.
error.beforeCommand = true;
reject(error);
return;
}
src/main/update-service.cjs
+6 -2
View File
@@ -499,7 +499,9 @@ class UpdateService {
5000,
);
child.once?.("spawn", () => finish(resolve));
child.once?.("error", (error) => finish(reject, error));
// Kept attached rather than `once`: a process that fails to start can
// report a second error, and an unhandled 'error' event ends this process.
child.on?.("error", (error) => finish(reject, error));
if (!child.once) finish(resolve);
});
@@ -630,7 +632,9 @@ class UpdateService {
clearTimeout(timer);
resolve();
});
child.once?.("error", (error) => {
// Kept attached rather than `once`: a second error would otherwise have no
// listener left, and an unhandled 'error' event ends this process.
child.on?.("error", (error) => {
clearTimeout(timer);
reject(error);
});
tests/ssh-connection-pool.test.mjs
+255
View File
@@ -0,0 +1,255 @@
import test from "node:test";
import assert from "node:assert/strict";
import { EventEmitter } from "node:events";
import { createRequire } from "node:module";
const require = createRequire(import.meta.url);
const ssh2Path = require.resolve("ssh2");
const realSsh2 = require("ssh2");
// SshService resolves ssh2 lazily and after an await, so the replacement has to
// stay in place until the whole operation settles.
async function withFakeSsh2(Client, operation) {
require.cache[ssh2Path] = { id: ssh2Path, filename: ssh2Path, loaded: true, exports: { ...realSsh2, Client } };
try {
return await operation();
} finally {
require.cache[ssh2Path] = { id: ssh2Path, filename: ssh2Path, loaded: true, exports: realSsh2 };
}
}
const { SshService } = require("../src/main/ssh-service.cjs");
function makeStore(overrides = {}) {
const server = {
id: "unraid",
host: "tower",
port: 22,
username: "root",
authType: "password",
basePath: "/mnt/user/appdata",
hostFingerprint: "SHA256:trusted",
...overrides,
};
return {
server,
getServer: () => server,
getServerCredentials: () => ({ password: "secret", passphrase: "" }),
};
}
// A client that reports what the pool does to it: how often it connected, how
// many channels it opened, and whether it was closed.
function fakeClientFactory({ execBehaviour = () => ({ ok: true }) } = {}) {
const state = { connects: 0, execs: 0, ends: 0, instances: [] };
class FakeClient extends EventEmitter {
constructor() {
super();
this.ended = false;
state.instances.push(this);
}
connect(options) {
state.connects += 1;
options.hostVerifier(Buffer.from("host key"));
setImmediate(() => this.emit("ready"));
}
exec(command, callback) {
state.execs += 1;
const outcome = execBehaviour(state.execs, this);
if (outcome.channelError) {
setImmediate(() => callback(outcome.channelError));
return;
}
const stream = new EventEmitter();
stream.stderr = new EventEmitter();
// A channel that closes without an exit status reports null, which is how
// a connection lost mid-command surfaces. That is not the same as 0.
const closeCode = Object.hasOwn(outcome, "exitCode") ? outcome.exitCode : 0;
setImmediate(() => {
stream.emit("data", Buffer.from(outcome.stdout ?? "ok"));
stream.emit("close", closeCode, null);
});
callback(null, stream);
}
end() {
if (this.ended) return;
this.ended = true;
state.ends += 1;
setImmediate(() => this.emit("close"));
}
}
return { FakeClient, state };
}
function service(store, options = {}) {
return new SshService({ store, diagnostics: null, ...options });
}
const run = withFakeSsh2;
test("a sequence of commands to one server shares a single connection", async () => {
const { FakeClient, state } = fakeClientFactory();
const store = makeStore();
const ssh = service(store);
for (let index = 0; index < 5; index += 1) {
await run(FakeClient, () => ssh.exec("unraid", `echo ${index}`));
}
assert.equal(state.execs, 5);
assert.equal(state.connects, 1, "five commands, one handshake");
ssh.closeAll();
});
test("concurrent commands share the connection and it survives until the last one finishes", async () => {
const { FakeClient, state } = fakeClientFactory();
const ssh = service(makeStore());
await run(FakeClient, () => Promise.all([
ssh.exec("unraid", "one"),
ssh.exec("unraid", "two"),
ssh.exec("unraid", "three"),
]));
assert.equal(state.connects, 1);
assert.equal(state.execs, 3);
assert.equal(state.ends, 0, "the shared connection is not closed while it is idle in the pool");
ssh.closeAll();
assert.equal(state.ends, 1);
});
test("a connection that died while pooled is replaced and the command runs once", async () => {
const { FakeClient, state } = fakeClientFactory({
execBehaviour: (call, client) => (call === 2 && !client.reopened
? { channelError: Object.assign(new Error("channel open failure"), { code: "ERR_CHANNEL" }) }
: { ok: true }),
});
const ssh = service(makeStore());
await run(FakeClient, () => ssh.exec("unraid", "first"));
const result = await run(FakeClient, () => ssh.exec("unraid", "second"));
assert.equal(result.stdout, "ok");
assert.equal(state.connects, 2, "the stale connection is replaced");
assert.equal(state.execs, 3, "the failed attempt never reached the server, so it is retried once");
ssh.closeAll();
});
test("a command that reached the server is never retried, not even on a reused connection", async () => {
let deployAttempts = 0;
const { FakeClient, state } = fakeClientFactory({
execBehaviour: (call) => {
if (call === 1) return { ok: true };
deployAttempts += 1;
return { exitCode: 1, stdout: "docker compose failed" };
},
});
const ssh = service(makeStore());
// The first command establishes the pooled connection, so the deployment below
// runs on a reused one - the case where a retry would be tempting.
await run(FakeClient, () => ssh.exec("unraid", "true"));
await assert.rejects(() => run(FakeClient, () => ssh.exec("unraid", "docker compose up -d")), (error) => {
assert.equal(error.code, "SSH_COMMAND_FAILED");
return true;
});
assert.equal(deployAttempts, 1, "a deployment command is never repeated by the pool");
assert.equal(state.connects, 1);
ssh.closeAll();
});
test("a connection lost while a command was running is not retried either", async () => {
let attempts = 0;
const { FakeClient, state } = fakeClientFactory({
execBehaviour: (call) => {
if (call === 1) return { ok: true };
attempts += 1;
// The stream opened, so the server may already be acting on this command.
return { exitCode: null, stdout: "" };
},
});
const ssh = service(makeStore());
await run(FakeClient, () => ssh.exec("unraid", "true"));
await assert.rejects(() => run(FakeClient, () => ssh.exec("unraid", "docker compose up -d")), (error) => {
assert.equal(error.code, "SSH_COMMAND_FAILED");
return true;
});
assert.equal(attempts, 1);
assert.equal(state.connects, 1);
ssh.closeAll();
});
test("a first connection that cannot be established is reported without a retry", async () => {
class RefusingClient extends EventEmitter {
connect() {
setImmediate(() => this.emit("error", Object.assign(new Error("ECONNREFUSED"), { code: "ECONNREFUSED" })));
}
end() {}
}
const ssh = service(makeStore());
await assert.rejects(() => run(RefusingClient, () => ssh.exec("unraid", "true")), /SSH connection failed/);
assert.equal(ssh.sessions.size, 0, "a failed connection is not pooled");
});
test("a trust-on-first-use connection is never pooled or reused", async () => {
const { FakeClient, state } = fakeClientFactory();
const ssh = service(makeStore({ hostFingerprint: "" }));
await run(FakeClient, () => ssh.test("unraid", { trustOnFirstUse: true }));
await run(FakeClient, () => ssh.test("unraid", { trustOnFirstUse: true }));
assert.equal(state.connects, 2, "an unverified connection is opened fresh every time");
assert.equal(ssh.sessions.size, 0);
assert.equal(state.ends, 2, "and closed immediately afterwards");
});
test("changing the server identity or credentials invalidates the pooled connection", async () => {
const { FakeClient, state } = fakeClientFactory();
const store = makeStore();
const ssh = service(store);
await run(FakeClient, () => ssh.exec("unraid", "before"));
assert.equal(state.connects, 1);
store.server.hostFingerprint = "SHA256:rotated";
await run(FakeClient, () => ssh.exec("unraid", "after"));
assert.equal(state.connects, 2, "the previous connection is not reused across an identity change");
ssh.closeAll();
});
test("an idle connection is closed after its lifetime and reopened on demand", async () => {
const { FakeClient, state } = fakeClientFactory();
const ssh = service(makeStore(), { idleConnectionMs: 40 });
await run(FakeClient, () => ssh.exec("unraid", "one"));
assert.equal(state.ends, 0);
await new Promise((resolve) => setTimeout(resolve, 120));
assert.equal(state.ends, 1, "the idle connection is released");
assert.equal(ssh.sessions.size, 0);
await run(FakeClient, () => ssh.exec("unraid", "two"));
assert.equal(state.connects, 2);
ssh.closeAll();
});
test("an error on an idle pooled connection is absorbed instead of terminating the process", async () => {
const { FakeClient, state } = fakeClientFactory();
const ssh = service(makeStore());
await run(FakeClient, () => ssh.exec("unraid", "one"));
const pooled = state.instances.at(-1);
pooled.emit("error", new Error("read ECONNRESET"));
await new Promise((resolve) => setTimeout(resolve, 20));
assert.equal(ssh.sessions.size, 0, "the dead connection leaves the pool");
await run(FakeClient, () => ssh.exec("unraid", "two"));
assert.equal(state.connects, 2);
ssh.closeAll();
});