diff --git a/SOURCE_MANIFEST.txt b/SOURCE_MANIFEST.txt index 3e00822..cc2294e 100644 --- a/SOURCE_MANIFEST.txt +++ b/SOURCE_MANIFEST.txt @@ -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 diff --git a/main.cjs b/main.cjs index 7f39ef6..5ad4778 100644 --- a/main.cjs +++ b/main.cjs @@ -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", {})) diff --git a/reports/architecture-audit.json b/reports/architecture-audit.json index 57a4cb8..6f001af 100644 --- a/reports/architecture-audit.json +++ b/reports/architecture-audit.json @@ -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, diff --git a/reports/architecture-audit.md b/reports/architecture-audit.md index 64ab467..0de4932 100644 --- a/reports/architecture-audit.md +++ b/reports/architecture-audit.md @@ -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 diff --git a/src/main/diagnostics-service.cjs b/src/main/diagnostics-service.cjs index 49bc130..6167076 100644 --- a/src/main/diagnostics-service.cjs +++ b/src/main/diagnostics-service.cjs @@ -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); } diff --git a/src/main/ssh-service.cjs b/src/main/ssh-service.cjs index 37b31de..624678e 100644 --- a/src/main/ssh-service.cjs +++ b/src/main/ssh-service.cjs @@ -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; } diff --git a/src/main/update-service.cjs b/src/main/update-service.cjs index 4dd30e2..bcff3db 100644 --- a/src/main/update-service.cjs +++ b/src/main/update-service.cjs @@ -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); }); diff --git a/tests/ssh-connection-pool.test.mjs b/tests/ssh-connection-pool.test.mjs new file mode 100644 index 0000000..6c2a902 --- /dev/null +++ b/tests/ssh-connection-pool.test.mjs @@ -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(); +});