diff --git a/apps/dokploy/__test__/server/exec-async-remote-lifecycle.test.ts b/apps/dokploy/__test__/server/exec-async-remote-lifecycle.test.ts new file mode 100644 index 000000000..970a1f484 --- /dev/null +++ b/apps/dokploy/__test__/server/exec-async-remote-lifecycle.test.ts @@ -0,0 +1,211 @@ +import { type ChildProcess, execFileSync, spawn } from "node:child_process"; +import { findServerById } from "@dokploy/server/services/server"; +import { + ExecError, + execAsyncRemote, +} from "@dokploy/server/utils/process/execAsync"; +import { type Connection, Server, utils } from "ssh2"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@dokploy/server/services/server", () => ({ findServerById: vi.fn() })); + +const alive = (pid: number) => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +}; + +describe.skipIf(process.platform === "win32")( + "real loopback SSH process lifecycle", + () => { + const key = utils.generateKeyPairSync("ed25519"); + let server: Server; + const connections: Connection[] = []; + const processes: ChildProcess[] = []; + const observedPids = new Set(); + + const observeGroup = (group: number) => { + for (const line of execFileSync("ps", ["-axo", "pid=,pgid="], { + encoding: "utf8", + }).split("\n")) { + const [pid, pgid] = line.trim().split(/\s+/).map(Number); + if (pid && pgid === group) observedPids.add(pid); + } + }; + + beforeEach(async () => { + observedPids.clear(); + server = new Server({ hostKeys: [key.private] }, (connection) => { + connections.push(connection); + connection.on("error", () => {}); + connection.on("authentication", (context) => context.accept()); + connection.on("ready", () => + connection.on("session", (accept) => { + const session = accept(); + // Ignore SSH signals deliberately: the remote deadline must survive transport loss. + session.on("signal", () => {}); + session.on("exec", (acceptExec, _reject, info) => { + const channel = acceptExec(); + const child = spawn("/bin/sh", ["-c", info.command], { + detached: true, + stdio: ["ignore", "pipe", "pipe"], + }); + processes.push(child); + if (child.pid) observedPids.add(child.pid); + child.stdout?.on("data", (chunk: Buffer) => { + if (child.pid) observeGroup(child.pid); + if (!channel.destroyed) channel.write(chunk); + }); + child.stderr?.on("data", (chunk: Buffer) => { + if (!channel.destroyed) channel.stderr.write(chunk); + }); + child.on("close", (code, signal) => { + if (channel.destroyed) return; + if (signal) channel.exit(signal); + else channel.exit(code ?? 1); + channel.end(); + }); + }); + }), + ); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") + throw new Error("Missing fixture port"); + vi.mocked(findServerById).mockResolvedValue({ + sshKeyId: "fixture", + ipAddress: "127.0.0.1", + port: address.port, + username: "fixture", + sshKey: { privateKey: key.private }, + } as never); + }); + + afterEach(async () => { + for (const child of processes.splice(0)) { + if (!child.pid) continue; + try { + process.kill(-child.pid, "SIGKILL"); + } catch { + /* already gone */ + } + } + for (const connection of connections.splice(0)) connection.end(); + await new Promise((resolve) => server.close(() => resolve())); + }); + + it("kills the remote shell, child, grandchild and watchdog after timeout even when signals are ignored", async () => { + const command = + "echo shell=$$; sh -c 'echo child=$$; sleep 30 & echo grandchild=$!; echo partial >&2; wait' & wait"; + const started = Date.now(); + await expect( + execAsyncRemote("fixture", command, undefined, { timeout: 2000 }), + ).rejects.toSatisfy( + (error: unknown) => + error instanceof ExecError && + /timed out/.test(error.message) && + (error.stdout?.includes("grandchild=") ?? false) && + (error.stderr?.includes("partial") ?? false), + ); + expect(Date.now() - started).toBeLessThan(3500); + expect(observedPids.size).toBeGreaterThanOrEqual(3); + await vi.waitFor( + () => { + expect([...observedPids].filter(alive)).toEqual([]); + }, + { timeout: 2000, interval: 30 }, + ); + }); + + it("retains remote cleanup after the SSH transport disconnects", async () => { + let disconnected = false; + const pending = execAsyncRemote( + "fixture", + "echo shell=$$; sleep 30 & echo grandchild=$!; wait", + () => { + if (disconnected) return; + disconnected = true; + connections[0]?.end(); + }, + { timeout: 2000 }, + ); + await expect(pending).rejects.toBeInstanceOf(ExecError); + expect(observedPids.size).toBeGreaterThanOrEqual(3); + await vi.waitFor( + () => expect([...observedPids].filter(alive)).toEqual([]), + { timeout: 2000, interval: 30 }, + ); + }); + + it.each(["1", "2"])( + "retains the deadline while an orphan grandchild holds output fd %s", + async (fd) => { + await expect( + execAsyncRemote( + "fixture", + `sleep 30 >&${fd} & echo grandchild=$!`, + undefined, + { timeout: 2000 }, + ), + ).rejects.toSatisfy( + (error: unknown) => + error instanceof ExecError && /timed out/.test(error.message), + ); + await vi.waitFor( + () => expect([...observedPids].filter(alive)).toEqual([]), + { timeout: 2000, interval: 30 }, + ); + }, + ); + + it.each(["exit 0", "exec sh -c 'exit 0'"])( + "does not leak the private exit status for %s", + async (command) => { + await expect( + execAsyncRemote("fixture", command, undefined, { timeout: 2000 }), + ).resolves.toEqual({ stdout: "", stderr: "" }); + }, + ); + + it("preserves large interleaved stdout and stderr independently", async () => { + const command = + "i=0; while [ $i -lt 2048 ]; do printf 'stdout\\n'; printf 'stderr\\n' >&2; i=$((i+1)); done"; + await expect( + execAsyncRemote("fixture", command, undefined, { timeout: 5000 }), + ).resolves.toEqual({ + stdout: "stdout\n".repeat(2048), + stderr: "stderr\n".repeat(2048), + }); + }); + + it.each([0, 7])( + "clears every watchdog process on normal exit %i", + async (code) => { + const pending = execAsyncRemote( + "fixture", + `echo shell=$$; sleep 0.1; exit ${code}`, + undefined, + { timeout: 30_000 }, + ); + if (code === 0) + await expect(pending).resolves.toMatchObject({ stderr: "" }); + else + await expect(pending).rejects.toMatchObject({ + exitCode: code, + stderr: "", + }); + await vi.waitFor( + () => expect([...observedPids].filter(alive)).toEqual([]), + { timeout: 1000 }, + ); + }, + ); + }, +); diff --git a/apps/dokploy/__test__/server/exec-async-timeout.test.ts b/apps/dokploy/__test__/server/exec-async-timeout.test.ts index 3f30a40db..af30d20bb 100644 --- a/apps/dokploy/__test__/server/exec-async-timeout.test.ts +++ b/apps/dokploy/__test__/server/exec-async-timeout.test.ts @@ -11,6 +11,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const { ssh, MockClient } = vi.hoisted(() => { class MockStream { + signal = vi.fn(); close = vi.fn(() => { this.closed = true; }); @@ -27,14 +28,25 @@ const { ssh, MockClient } = vi.hoisted(() => { return this; } + stderrHandlers: Array<(...args: unknown[]) => void> = []; stderr = { - on: (_event: string, _cb: (...args: unknown[]) => void) => this, + on: (_event: string, cb: (...args: unknown[]) => void) => { + this.stderrHandlers.push(cb); + return this; + }, }; + emitStderr(data: string) { + for (const cb of this.stderrHandlers) cb(data); + } emitClose(code: number) { for (const cb of this.handlers.close ?? []) cb(code, null); } + emitError(error: Error) { + for (const cb of this.handlers.error ?? []) cb(error); + } + emitData(data: string) { for (const cb of this.handlers.data ?? []) cb(data); } @@ -45,6 +57,9 @@ const { ssh, MockClient } = vi.hoisted(() => { destroyed = false; execCommand = ""; execShouldFail: Error | null = null; + execShouldThrow: Error | null = null; + deferExec = false; + execCallback?: (err: Error | null, stream: MockStream) => void; stream = new MockStream(); end = vi.fn(() => { this.ended = true; @@ -69,11 +84,17 @@ const { ssh, MockClient } = vi.hoisted(() => { } connect() { + if (ssh.connectError) throw ssh.connectError; return this; } exec(command: string, cb: (err: Error | null, stream: MockStream) => void) { this.execCommand = command; + if (this.execShouldThrow) throw this.execShouldThrow; + if (this.deferExec) { + this.execCallback = cb; + return; + } if (this.execShouldFail) { cb(this.execShouldFail, this.stream); return; @@ -90,7 +111,10 @@ const { ssh, MockClient } = vi.hoisted(() => { } } - const ssh = { instances: [] as MockClient[] }; + const ssh = { + instances: [] as MockClient[], + connectError: null as Error | null, + }; return { ssh, MockClient }; }); @@ -173,6 +197,7 @@ describe("execAsyncRemote timeout", () => { beforeEach(() => { ssh.instances.length = 0; + ssh.connectError = null; vi.mocked(findServerById).mockReset(); vi.mocked(findServerById).mockResolvedValue(server as never); vi.useFakeTimers(); @@ -212,9 +237,10 @@ describe("execAsyncRemote timeout", () => { await flush(); const client = ssh.instances[0]; client?.emitReady(); - expect(client?.execCommand).toBe("nvidia-smi"); + expect(client?.execCommand).toContain("nvidia-smi"); await vi.advanceTimersByTimeAsync(1000); await rejected; + expect(client?.stream.signal).toHaveBeenCalledWith("KILL"); expect(client?.stream.close).toHaveBeenCalledTimes(1); expect(client?.end).toHaveBeenCalledTimes(1); client?.stream.emitClose(0); @@ -231,6 +257,7 @@ describe("execAsyncRemote timeout", () => { client?.stream.emitClose(0); await expect(pending).resolves.toEqual({ stdout: "", stderr: "" }); expect(vi.getTimerCount()).toBeLessThan(timersBeforeClose); + expect(vi.getTimerCount()).toBe(0); }); it("resolves on success and does not reject later when the timer would have fired", async () => { @@ -303,6 +330,103 @@ describe("execAsyncRemote timeout", () => { expect(client?.end).toHaveBeenCalled(); }); + it("cleans timers and clients when connect throws synchronously", async () => { + ssh.connectError = new Error("invalid key"); + await expect(startRemote()).rejects.toMatchObject({ + message: "SSH connection failed", + }); + expect(vi.getTimerCount()).toBe(0); + expect(ssh.instances[0]?.destroy).toHaveBeenCalledOnce(); + }); + + it("cleans timers and clients when exec throws synchronously", async () => { + const pending = startRemote(); + const rejected = expect(pending).rejects.toMatchObject({ + message: "Remote command execution failed", + }); + await flush(); + const client = ssh.instances[0]; + if (client) client.execShouldThrow = new Error("channel unavailable"); + client?.emitReady(); + await rejected; + expect(vi.getTimerCount()).toBe(0); + expect(client?.destroy).toHaveBeenCalledOnce(); + }); + + it("closes a late channel even when its signal method throws", async () => { + const pending = startRemote(); + const rejected = expect(pending).rejects.toBeInstanceOf(ExecError); + await flush(); + const client = ssh.instances[0]; + if (client) client.deferExec = true; + client?.emitReady(); + await vi.advanceTimersByTimeAsync(1000); + await rejected; + client?.stream.signal.mockImplementation(() => { + throw new Error("closed transport"); + }); + if (client) client.execCallback?.(null, client.stream); + expect(client?.stream.close).toHaveBeenCalledOnce(); + expect(client?.stream.destroy).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("preserves both partial streams and ignores late output and errors", async () => { + const onData = vi.fn(); + const pending = startRemote(onData); + const rejected = expect(pending).rejects.toMatchObject({ + stdout: "out", + stderr: "err", + }); + await flush(); + const client = ssh.instances[0]; + client?.emitReady(); + client?.stream.emitData("out"); + client?.stream.emitStderr("err"); + await vi.advanceTimersByTimeAsync(1000); + await rejected; + client?.stream.emitData("late out"); + client?.stream.emitStderr("late err"); + client?.emitError(new Error("late error")); + client?.stream.emitError(new Error("late stream error")); + client?.stream.emitClose(0); + expect(onData.mock.calls).toEqual([["out"], ["err"]]); + expect(client?.end).toHaveBeenCalledTimes(1); + expect(client?.destroy).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + + it.each([undefined, 0, -1, Number.NaN, Number.POSITIVE_INFINITY])( + "does not wrap commands without a positive finite timeout: %s", + async (timeout) => { + const pending = execAsyncRemote( + "server-1", + "original command", + undefined, + { timeout }, + ); + await flush(); + const client = ssh.instances[0]; + client?.emitReady(); + expect(client?.execCommand).toBe("original command"); + expect(vi.getTimerCount()).toBe(0); + client?.stream.emitClose(0); + await pending; + expect(client?.stream.signal).not.toHaveBeenCalled(); + }, + ); + + it("keeps a normal nonzero exit distinct from timeout", async () => { + const pending = startRemote(); + const rejected = expect(pending).rejects.toMatchObject({ exitCode: 7 }); + await flush(); + const client = ssh.instances[0]; + client?.emitReady(); + client?.stream.emitClose(7); + await rejected; + expect(vi.getTimerCount()).toBe(0); + }); + it("still delivers onData chunks if the command hangs afterward", async () => { const chunks: string[] = []; const pending = startRemote((data) => { diff --git a/packages/server/src/utils/process/execAsync.ts b/packages/server/src/utils/process/execAsync.ts index 9696d2aa6..9b4a8c8ae 100644 --- a/packages/server/src/utils/process/execAsync.ts +++ b/packages/server/src/utils/process/execAsync.ts @@ -1,8 +1,9 @@ import { type ChildProcess, exec, execFile, spawn } from "node:child_process"; import util from "node:util"; import { findServerById } from "@dokploy/server/services/server"; -import { Client } from "ssh2"; +import { Client, type ClientChannel } from "ssh2"; import { ExecError } from "./ExecError"; +import { withRemoteDeadline } from "./remote-deadline"; export class WriteFileRemoteError extends Error { constructor( @@ -129,7 +130,7 @@ export const execAsync = async ( options?: ExecAsyncOptions, ): Promise<{ stdout: string; stderr: string }> => { const timeout = options?.timeout; - if (timeout != null && timeout > 0) { + if (timeout != null && Number.isFinite(timeout) && timeout > 0) { return execAsyncTimed(command, { ...options, timeout }); } @@ -271,13 +272,40 @@ export const execAsyncRemote = async ( let stdout = ""; let stderr = ""; - const timeoutMs = options?.timeout; + const requestedTimeout = options?.timeout; + const timeoutMs = + requestedTimeout != null && + Number.isFinite(requestedTimeout) && + requestedTimeout > 0 + ? requestedTimeout + : undefined; + const started = Date.now(); return new Promise((resolve, reject) => { const conn = new Client(); - let stream: { close: () => void; destroy: () => void } | undefined; + let stream: ClientChannel | undefined; let settled = false; let timer: ReturnType | undefined; + const closeStream = (channel?: ClientChannel) => { + if (timeoutMs !== undefined) { + try { + channel?.signal("KILL"); + } catch { + /* remote watchdog owns cleanup */ + } + } + try { + channel?.close(); + } catch { + /* already closed */ + } + try { + channel?.destroy(); + } catch { + /* already destroyed */ + } + }; + const settle = ( error?: ExecError, value?: { stdout: string; stderr: string }, @@ -289,16 +317,7 @@ export const execAsyncRemote = async ( timer = undefined; } if (error) { - try { - stream?.close(); - } catch { - // already closed - } - try { - stream?.destroy(); - } catch { - // already destroyed - } + closeStream(stream); try { conn.end(); } catch { @@ -333,110 +352,158 @@ export const execAsyncRemote = async ( }, timeoutMs); } - sleep(1000); - conn - .once("ready", () => { - if (settled) return; - conn.exec(command, (err, commandStream) => { - if (settled) { - try { - commandStream?.close(); - } catch { - // already closed - } - return; - } - if (err) { - onData?.(err.message); + try { + conn + .once("ready", () => { + if (settled) return; + const remoteCommand = + timeoutMs === undefined + ? command + : withRemoteDeadline( + command, + Math.max(1, timeoutMs - (Date.now() - started)), + ); + try { + conn.exec(remoteCommand, (err, commandStream) => { + if (settled) { + closeStream(commandStream); + return; + } + if (err) { + onData?.(err.message); + settle( + new ExecError( + `Remote command execution failed: ${err.message}`, + { + command, + serverId, + originalError: err, + }, + ), + ); + return; + } + stream = commandStream; + commandStream + .on("error", (error: Error) => { + if (settled) return; + settle( + new ExecError( + `Remote command stream failed: ${error.message}`, + { + command, + stdout, + stderr, + serverId, + originalError: error, + }, + ), + ); + }) + .on("close", (code: number, _signal: string) => { + if (code === 0) { + settle(undefined, { stdout, stderr }); + } else { + settle( + new ExecError( + `Remote command failed with exit code ${code}`, + { + command, + stdout, + stderr, + exitCode: code, + serverId, + }, + ), + ); + } + }) + .on("data", (data: string) => { + if (settled) return; + stdout += data.toString(); + onData?.(data.toString()); + }) + .stderr.on("data", (data) => { + if (settled) return; + stderr += data.toString(); + onData?.(data.toString()); + }); + }); + } catch (error) { settle( - new ExecError(`Remote command execution failed: ${err.message}`, { + new ExecError("Remote command execution failed", { + command, + stdout, + stderr, + serverId, + originalError: error instanceof Error ? error : undefined, + }), + ); + } + }) + .on("error", (err) => { + if (settled) return; + if (err.level === "client-authentication") { + const technicalDetail = `Error: ${err.message} ${err.level}`; + const friendlyMessage = [ + "", + "❌ Couldn't connect to your server — the SSH key was not accepted.", + "", + "This usually means the key doesn't match what's on the server, or the key format is invalid.", + "", + `Technical details: ${technicalDetail}`, + "", + "💡 Hints:", + " • Check that the SSH key you added in Dokploy is the same one installed on the server (e.g. in ~/.ssh/authorized_keys).", + " • Try generating a new SSH key in Dokploy and add only the public key to the server, then try again.", + " • Make sure to follow the instructions on the Setup Server Button on the SSH Keys tab and then click on deployments tab and check the logs for more details.", + ].join("\n"); + onData?.(friendlyMessage); + settle( + new ExecError( + `Authentication failed: Invalid SSH private key. ${friendlyMessage}`, + { + command, + serverId, + originalError: err, + }, + ), + ); + } else { + const errorMsg = `SSH connection error: ${err.message}`; + onData?.(errorMsg); + settle( + new ExecError(errorMsg, { + stdout, + stderr, command, serverId, originalError: err, }), ); - return; } - stream = commandStream; - commandStream - .on("close", (code: number, _signal: string) => { - if (code === 0) { - settle(undefined, { stdout, stderr }); - } else { - settle( - new ExecError( - `Remote command failed with exit code ${code}`, - { - command, - stdout, - stderr, - exitCode: code, - serverId, - }, - ), - ); - } - }) - .on("data", (data: string) => { - stdout += data.toString(); - onData?.(data.toString()); - }) - .stderr.on("data", (data) => { - stderr += data.toString(); - onData?.(data.toString()); - }); + }) + .connect({ + host: server.ipAddress, + port: server.port, + username: server.username, + privateKey: server.sshKey?.privateKey, + timeout: timeoutMs != null && timeoutMs > 0 ? timeoutMs : 99999, + ...(timeoutMs != null && timeoutMs > 0 + ? { readyTimeout: timeoutMs } + : {}), }); - }) - .on("error", (err) => { - if (settled) return; - if (err.level === "client-authentication") { - const technicalDetail = `Error: ${err.message} ${err.level}`; - const friendlyMessage = [ - "", - "❌ Couldn't connect to your server — the SSH key was not accepted.", - "", - "This usually means the key doesn't match what's on the server, or the key format is invalid.", - "", - `Technical details: ${technicalDetail}`, - "", - "💡 Hints:", - " • Check that the SSH key you added in Dokploy is the same one installed on the server (e.g. in ~/.ssh/authorized_keys).", - " • Try generating a new SSH key in Dokploy and add only the public key to the server, then try again.", - " • Make sure to follow the instructions on the Setup Server Button on the SSH Keys tab and then click on deployments tab and check the logs for more details.", - ].join("\n"); - onData?.(friendlyMessage); - settle( - new ExecError( - `Authentication failed: Invalid SSH private key. ${friendlyMessage}`, - { - command, - serverId, - originalError: err, - }, - ), - ); - } else { - const errorMsg = `SSH connection error: ${err.message}`; - onData?.(errorMsg); - settle( - new ExecError(errorMsg, { - command, - serverId, - originalError: err, - }), - ); - } - }) - .connect({ - host: server.ipAddress, - port: server.port, - username: server.username, - privateKey: server.sshKey?.privateKey, - timeout: timeoutMs != null && timeoutMs > 0 ? timeoutMs : 99999, - ...(timeoutMs != null && timeoutMs > 0 - ? { readyTimeout: timeoutMs } - : {}), - }); + } catch (error) { + settle( + new ExecError("SSH connection failed", { + command, + stdout, + stderr, + serverId, + originalError: error instanceof Error ? error : undefined, + }), + ); + } }); }; diff --git a/packages/server/src/utils/process/remote-deadline.ts b/packages/server/src/utils/process/remote-deadline.ts new file mode 100644 index 000000000..13cc98de2 --- /dev/null +++ b/packages/server/src/utils/process/remote-deadline.ts @@ -0,0 +1,39 @@ +const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`; + +/** OpenSSH gives each non-PTY command its own process group, including its children. */ +export const withRemoteDeadline = ( + command: string, + timeout: number, +): string => { + const seconds = timeout / 1000; + // The status pipe is private; output relays keep the deadline alive until inherited pipes drain. + return `exec sh -c ${quote(` +kill -0 -$$ 2>/dev/null || { echo 'Timed SSH command requires an isolated process group' >&2; exit 125; } +command -v cat >/dev/null 2>&1 || { echo 'Timed SSH command requires cat' >&2; exit 125; } +if sleep 0.001 2>/dev/null; then + duration=${seconds} +elif sleep 0 2>/dev/null; then + duration=${Math.ceil(seconds)} +else + echo 'Timed SSH command requires sleep' >&2 + exit 125 +fi +sleep "$duration" & +deadline=$! +( + result=$( + exec 3>&1 + { + { sh -c ${quote(command)} 3>&- 4>&- 5>&-; printf '%s' "$?" >&3; } | cat >&4 + } 2>&1 | cat >&5 + ) + kill "$deadline" 2>/dev/null + exit "$result" +) 4>&1 5>&2 & +command_pid=$! +if wait "$deadline" 2>/dev/null; then + kill -s KILL 0 +fi +wait "$command_pid" +`)}`; +};