fix(update): add a timeout for deployment/rebuild and add a cancel button.
This commit is contained in:
1 parent
c30f828d2c
commit
81622b5a4c
8 files changed
+443
-19
No files matched your search
@@ -234,6 +234,8 @@ export default function DeployContainerModal({
|
||||
const [networksLoading, setNetworksLoading] = useState(false);
|
||||
const [networksError, setNetworksError] = useState("");
|
||||
const envFileInputRef = useRef<HTMLInputElement | null>(null);
|
||||
const currentJobIdRef = useRef<string | null>(null);
|
||||
const cancelRequestedRef = useRef(false);
|
||||
const [uploadTargetEnvId, setUploadTargetEnvId] = useState<string | null>(
|
||||
null,
|
||||
);
|
||||
@@ -779,6 +781,8 @@ export default function DeployContainerModal({
|
||||
event.preventDefault();
|
||||
setError("");
|
||||
setLoading(true);
|
||||
currentJobIdRef.current = null;
|
||||
cancelRequestedRef.current = false;
|
||||
let processLogLines = [
|
||||
`[deploy] Starting deployment for ${form.name || "new container"}`,
|
||||
];
|
||||
@@ -946,6 +950,7 @@ export default function DeployContainerModal({
|
||||
|
||||
const jobResponse = await containersApi.createDeployJob(payload);
|
||||
const job = jobResponse.data;
|
||||
currentJobIdRef.current = job.id;
|
||||
let finalStatus = job.status;
|
||||
let finalError = job.error;
|
||||
|
||||
@@ -961,6 +966,33 @@ export default function DeployContainerModal({
|
||||
}),
|
||||
);
|
||||
|
||||
onProcessUpdate?.({
|
||||
cancelAction: {
|
||||
label: "Cancel",
|
||||
loadingLabel: "Cancelling",
|
||||
onClick: () => {
|
||||
const jobId = currentJobIdRef.current;
|
||||
if (!jobId || cancelRequestedRef.current) return;
|
||||
|
||||
cancelRequestedRef.current = true;
|
||||
onProcessUpdate?.({
|
||||
statusLabel: "Cancelling",
|
||||
cancelAction: {
|
||||
label: "Cancel",
|
||||
loadingLabel: "Cancelling",
|
||||
isLoading: true,
|
||||
disabled: true,
|
||||
onClick: () => undefined,
|
||||
},
|
||||
});
|
||||
|
||||
void containersApi
|
||||
.cancelJob(jobId)
|
||||
.catch(() => undefined);
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
await containersApi.streamJob(job.id, {
|
||||
onLog: (entry) => {
|
||||
updateProcessTerminal(
|
||||
@@ -977,10 +1009,25 @@ export default function DeployContainerModal({
|
||||
},
|
||||
});
|
||||
|
||||
if (cancelRequestedRef.current && finalStatus !== "cancelled") {
|
||||
const cancelledJob = currentJobIdRef.current
|
||||
? await containersApi.getJob(currentJobIdRef.current)
|
||||
: null;
|
||||
finalStatus = cancelledJob?.data.status ?? "cancelled";
|
||||
finalError =
|
||||
cancelledJob?.data.error ??
|
||||
cancelledJob?.data.cancelReason ??
|
||||
"Deployment cancelled";
|
||||
}
|
||||
|
||||
if (finalStatus === "error") {
|
||||
throw new Error(finalError || "Deployment job failed");
|
||||
}
|
||||
|
||||
if (finalStatus === "cancelled") {
|
||||
throw new Error(finalError || "Deployment cancelled");
|
||||
}
|
||||
|
||||
updateDeployTimelineModal({
|
||||
onProcessUpdate,
|
||||
activeStep: 3,
|
||||
@@ -1003,6 +1050,7 @@ export default function DeployContainerModal({
|
||||
"[deploy] Done",
|
||||
].filter(Boolean),
|
||||
});
|
||||
onProcessUpdate?.({ cancelAction: undefined });
|
||||
|
||||
onClose();
|
||||
} catch (err: unknown) {
|
||||
@@ -1013,7 +1061,9 @@ export default function DeployContainerModal({
|
||||
name: form.name,
|
||||
message,
|
||||
});
|
||||
onProcessUpdate?.({ cancelAction: undefined });
|
||||
} finally {
|
||||
currentJobIdRef.current = null;
|
||||
setLoading(false);
|
||||
}
|
||||
};
|
||||
|
||||
+48
@@ -424,6 +424,8 @@ export default function AppContainerDetailPage() {
|
||||
closeProcessLogs,
|
||||
} = useProcessLogsModal();
|
||||
const containerRecordRef = useRef<Container | null>(null);
|
||||
const rebuildJobIdRef = useRef<string | null>(null);
|
||||
const rebuildCancelRequestedRef = useRef(false);
|
||||
const runtimeMetricsInFlightRef = useRef(false);
|
||||
const environmentContainersHref = `/projects/${params.projectId}/environments/${params.environmentId}`;
|
||||
const appDetailId = appDetail?.id;
|
||||
@@ -796,6 +798,8 @@ export default function AppContainerDetailPage() {
|
||||
setActing(true);
|
||||
setActiveAction("rebuild");
|
||||
setActionError("");
|
||||
rebuildJobIdRef.current = null;
|
||||
rebuildCancelRequestedRef.current = false;
|
||||
const baseTerminalLines = [
|
||||
`[rebuild] Starting rebuild for ${containerRecord.name}`,
|
||||
`[rebuild] Source: ${containerRecord.sourceType ?? "unknown"}`,
|
||||
@@ -841,6 +845,7 @@ export default function AppContainerDetailPage() {
|
||||
params.containerId,
|
||||
);
|
||||
const job = jobResponse.data;
|
||||
rebuildJobIdRef.current = job.id;
|
||||
let finalStatus = job.status;
|
||||
let finalError = job.error;
|
||||
|
||||
@@ -856,6 +861,31 @@ export default function AppContainerDetailPage() {
|
||||
},
|
||||
);
|
||||
|
||||
updateProcessLogs({
|
||||
cancelAction: {
|
||||
label: "Cancel",
|
||||
loadingLabel: "Cancelling",
|
||||
onClick: () => {
|
||||
const jobId = rebuildJobIdRef.current;
|
||||
if (!jobId || rebuildCancelRequestedRef.current) return;
|
||||
|
||||
rebuildCancelRequestedRef.current = true;
|
||||
updateProcessLogs({
|
||||
statusLabel: "Cancelling",
|
||||
cancelAction: {
|
||||
label: "Cancel",
|
||||
loadingLabel: "Cancelling",
|
||||
isLoading: true,
|
||||
disabled: true,
|
||||
onClick: () => undefined,
|
||||
},
|
||||
});
|
||||
|
||||
void containersApi.cancelJob(jobId).catch(() => undefined);
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
await containersApi.streamJob(job.id, {
|
||||
onLog: (entry) => {
|
||||
updateProcessTerminal([...latestTerminalLines, entry.message], {
|
||||
@@ -869,10 +899,25 @@ export default function AppContainerDetailPage() {
|
||||
},
|
||||
});
|
||||
|
||||
if (rebuildCancelRequestedRef.current && finalStatus !== "cancelled") {
|
||||
const cancelledJob = rebuildJobIdRef.current
|
||||
? await containersApi.getJob(rebuildJobIdRef.current)
|
||||
: null;
|
||||
finalStatus = cancelledJob?.data.status ?? "cancelled";
|
||||
finalError =
|
||||
cancelledJob?.data.error ??
|
||||
cancelledJob?.data.cancelReason ??
|
||||
"Rebuild cancelled";
|
||||
}
|
||||
|
||||
if (finalStatus === "error") {
|
||||
throw new Error(finalError || "Rebuild job failed");
|
||||
}
|
||||
|
||||
if (finalStatus === "cancelled") {
|
||||
throw new Error(finalError || "Rebuild cancelled");
|
||||
}
|
||||
|
||||
updateProcessLogs({
|
||||
timelineLogs: createRebuildTimeline(2),
|
||||
terminalLogs: [
|
||||
@@ -898,6 +943,7 @@ export default function AppContainerDetailPage() {
|
||||
].filter(Boolean),
|
||||
statusLabel: "100%",
|
||||
});
|
||||
updateProcessLogs({ cancelAction: undefined });
|
||||
} catch (rebuildError) {
|
||||
const message =
|
||||
rebuildError instanceof Error ? rebuildError.message : "Rebuild failed";
|
||||
@@ -911,7 +957,9 @@ export default function AppContainerDetailPage() {
|
||||
],
|
||||
statusLabel: "Failed",
|
||||
});
|
||||
updateProcessLogs({ cancelAction: undefined });
|
||||
} finally {
|
||||
rebuildJobIdRef.current = null;
|
||||
setActing(false);
|
||||
setActiveAction(null);
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
Circle,
|
||||
ImageIcon,
|
||||
Loader2,
|
||||
OctagonX,
|
||||
ScrollText,
|
||||
Terminal,
|
||||
X,
|
||||
@@ -45,6 +46,13 @@ export type ProcessLogsModalState = {
|
||||
imageSlot?: ReactNode;
|
||||
initialTab?: ProcessLogsModalTab;
|
||||
statusLabel?: string;
|
||||
cancelAction?: {
|
||||
label?: string;
|
||||
loadingLabel?: string;
|
||||
isLoading?: boolean;
|
||||
disabled?: boolean;
|
||||
onClick: () => void;
|
||||
};
|
||||
};
|
||||
|
||||
export type ProcessLogsModalTab = "timeline" | "terminal";
|
||||
@@ -157,6 +165,7 @@ export default function ProcessLogsModal({
|
||||
imageSlot,
|
||||
initialTab = "timeline",
|
||||
statusLabel,
|
||||
cancelAction,
|
||||
closeOnOverlayClick = true,
|
||||
onClose,
|
||||
}: ProcessLogsModalProps) {
|
||||
@@ -276,22 +285,59 @@ export default function ProcessLogsModal({
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<button
|
||||
type="button"
|
||||
className="btn btn-ghost"
|
||||
onClick={handleClose}
|
||||
aria-label="Close process logs modal"
|
||||
<div
|
||||
style={{
|
||||
width: 25,
|
||||
height: 25,
|
||||
padding: 0,
|
||||
display: "flex",
|
||||
alignItems: "center",
|
||||
gap: 8,
|
||||
flex: "0 0 auto",
|
||||
justifyContent: "center",
|
||||
color: "var(--text-muted)",
|
||||
}}
|
||||
>
|
||||
<X size={21} />
|
||||
</button>
|
||||
{cancelAction ? (
|
||||
<button
|
||||
type="button"
|
||||
className="btn btn-ghost"
|
||||
onClick={cancelAction.onClick}
|
||||
disabled={cancelAction.disabled || cancelAction.isLoading}
|
||||
aria-label={cancelAction.label ?? "Cancel process"}
|
||||
style={{
|
||||
minHeight: 30,
|
||||
padding: "0 10px",
|
||||
justifyContent: "center",
|
||||
color: "#ef4444",
|
||||
border: "1px solid rgba(239,68,68,0.35)",
|
||||
background: "rgba(239,68,68,0.08)",
|
||||
}}
|
||||
>
|
||||
{cancelAction.isLoading ? (
|
||||
<Loader2 size={14} className="animate-spin" />
|
||||
) : (
|
||||
<OctagonX size={14} />
|
||||
)}
|
||||
{cancelAction.isLoading
|
||||
? (cancelAction.loadingLabel ?? "Cancelling")
|
||||
: (cancelAction.label ?? "Cancel")}
|
||||
</button>
|
||||
) : null}
|
||||
{!cancelAction ? (
|
||||
<button
|
||||
type="button"
|
||||
className="btn btn-ghost"
|
||||
onClick={handleClose}
|
||||
aria-label="Close process logs modal"
|
||||
style={{
|
||||
width: 25,
|
||||
height: 25,
|
||||
padding: 0,
|
||||
flex: "0 0 auto",
|
||||
justifyContent: "center",
|
||||
color: "var(--text-muted)",
|
||||
}}
|
||||
>
|
||||
<X size={21} />
|
||||
</button>
|
||||
) : null}
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div
|
||||
|
||||
+14
-1
@@ -1561,12 +1561,19 @@ export interface ProcessJobLogEntry {
|
||||
export interface ProcessJobRecord {
|
||||
id: string;
|
||||
type: string;
|
||||
status: "queued" | "running" | "success" | "error";
|
||||
status:
|
||||
| "queued"
|
||||
| "running"
|
||||
| "cancelling"
|
||||
| "cancelled"
|
||||
| "success"
|
||||
| "error";
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
logs: ProcessJobLogEntry[];
|
||||
result?: unknown;
|
||||
error?: string;
|
||||
cancelReason?: string;
|
||||
}
|
||||
|
||||
function parseServerSentEventBlock(block: string) {
|
||||
@@ -1622,6 +1629,12 @@ export const containers = {
|
||||
get<{ success: boolean; data: ProcessJobRecord }>(
|
||||
`/containers/jobs/${jobId}`,
|
||||
),
|
||||
cancelJob: (jobId: string) =>
|
||||
post<{ success: boolean; data: ProcessJobRecord; message?: string }>(
|
||||
`/containers/jobs/${jobId}/cancel`,
|
||||
{},
|
||||
{ timeoutMs: 15000 },
|
||||
),
|
||||
streamJob: async (
|
||||
jobId: string,
|
||||
handlers: {
|
||||
|
||||
@@ -14,6 +14,7 @@ import { auditLog } from "../services/audit.service";
|
||||
import { dispatchRuntimeNotification } from "../services/notification.service";
|
||||
import {
|
||||
appendProcessJobLog,
|
||||
cancelProcessJob,
|
||||
createProcessJob,
|
||||
getProcessJob,
|
||||
serializeProcessJob,
|
||||
@@ -958,6 +959,10 @@ function buildInternalJobHeaders(headers: Record<string, unknown>) {
|
||||
return nextHeaders;
|
||||
}
|
||||
|
||||
function isProcessJobCancelling(job: { status: string }) {
|
||||
return job.status === "cancelling";
|
||||
}
|
||||
|
||||
function runInjectedContainerJob(input: {
|
||||
app: FastifyInstance;
|
||||
job: ReturnType<typeof createProcessJob>;
|
||||
@@ -968,7 +973,20 @@ function runInjectedContainerJob(input: {
|
||||
redactSecret?: string;
|
||||
}) {
|
||||
void Promise.resolve().then(async () => {
|
||||
updateProcessJob(input.job, { status: "running" });
|
||||
const abortController = new AbortController();
|
||||
if (isProcessJobCancelling(input.job)) {
|
||||
updateProcessJob(input.job, {
|
||||
status: "cancelled",
|
||||
error: input.job.cancelReason ?? "Job cancelled",
|
||||
});
|
||||
appendProcessJobLog(input.job, "[job] Cancelled");
|
||||
return;
|
||||
}
|
||||
|
||||
updateProcessJob(input.job, {
|
||||
status: "running",
|
||||
cancel: (reason) => abortController.abort(reason),
|
||||
});
|
||||
appendProcessJobLog(input.job, `[job] Started ${input.job.type}`);
|
||||
|
||||
try {
|
||||
@@ -987,8 +1005,21 @@ function runInjectedContainerJob(input: {
|
||||
headers: buildInternalJobHeaders(input.headers),
|
||||
payload: input.payload,
|
||||
}),
|
||||
{
|
||||
runIdPrefix: input.job.id,
|
||||
signal: abortController.signal,
|
||||
},
|
||||
);
|
||||
|
||||
if (abortController.signal.aborted || isProcessJobCancelling(input.job)) {
|
||||
updateProcessJob(input.job, {
|
||||
status: "cancelled",
|
||||
error: input.job.cancelReason ?? "Job cancelled",
|
||||
});
|
||||
appendProcessJobLog(input.job, "[job] Cancelled");
|
||||
return;
|
||||
}
|
||||
|
||||
const payload = response.json() as {
|
||||
error?: unknown;
|
||||
[key: string]: unknown;
|
||||
@@ -1005,6 +1036,19 @@ function runInjectedContainerJob(input: {
|
||||
appendProcessJobLog(input.job, "[job] Completed successfully");
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : "Job failed";
|
||||
if (
|
||||
abortController.signal.aborted ||
|
||||
isProcessJobCancelling(input.job) ||
|
||||
message === "Command cancelled"
|
||||
) {
|
||||
updateProcessJob(input.job, {
|
||||
status: "cancelled",
|
||||
error: input.job.cancelReason ?? "Job cancelled",
|
||||
});
|
||||
appendProcessJobLog(input.job, "[job] Cancelled");
|
||||
return;
|
||||
}
|
||||
|
||||
updateProcessJob(input.job, { status: "error", error: message });
|
||||
appendProcessJobLog(input.job, `[job] Failed: ${message}`);
|
||||
}
|
||||
@@ -1043,6 +1087,33 @@ export async function containerRoutes(app: FastifyInstance) {
|
||||
},
|
||||
);
|
||||
|
||||
app.post(
|
||||
"/jobs/:jobId/cancel",
|
||||
{ preHandler: containerWriteAccess },
|
||||
async (req, reply) => {
|
||||
const { jobId } = req.params as { jobId: string };
|
||||
const job = getProcessJob(jobId);
|
||||
|
||||
if (
|
||||
!job ||
|
||||
job.userId !== req.userId ||
|
||||
job.organizationId !== req.organizationId
|
||||
) {
|
||||
return reply
|
||||
.status(404)
|
||||
.send({ success: false, error: "Job not found" });
|
||||
}
|
||||
|
||||
const cancelled = cancelProcessJob(job);
|
||||
|
||||
return reply.send({
|
||||
success: true,
|
||||
data: serializeProcessJob(job),
|
||||
message: cancelled ? "Cancellation requested" : "Job already finished",
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
app.get(
|
||||
"/jobs/:jobId/stream",
|
||||
{ preHandler: containerReadAccess },
|
||||
@@ -1087,7 +1158,11 @@ export async function containerRoutes(app: FastifyInstance) {
|
||||
});
|
||||
const statusInterval = setInterval(() => {
|
||||
writeProcessJobStreamEvent(reply, "status", serializeProcessJob(job));
|
||||
if (job.status === "success" || job.status === "error") {
|
||||
if (
|
||||
job.status === "success" ||
|
||||
job.status === "error" ||
|
||||
job.status === "cancelled"
|
||||
) {
|
||||
clearInterval(statusInterval);
|
||||
unsubscribe();
|
||||
if (!reply.raw.destroyed && !reply.raw.writableEnded) {
|
||||
|
||||
@@ -1,6 +1,12 @@
|
||||
import { randomUUID } from "crypto";
|
||||
|
||||
export type ProcessJobStatus = "queued" | "running" | "success" | "error";
|
||||
export type ProcessJobStatus =
|
||||
| "queued"
|
||||
| "running"
|
||||
| "cancelling"
|
||||
| "cancelled"
|
||||
| "success"
|
||||
| "error";
|
||||
|
||||
export interface ProcessJobLogEntry {
|
||||
id: number;
|
||||
@@ -19,6 +25,8 @@ export interface ProcessJob {
|
||||
logs: ProcessJobLogEntry[];
|
||||
result?: unknown;
|
||||
error?: string;
|
||||
cancelReason?: string;
|
||||
cancel?: (reason?: string) => void;
|
||||
nextLogId: number;
|
||||
subscribers: Set<(entry: ProcessJobLogEntry) => void>;
|
||||
}
|
||||
@@ -31,7 +39,9 @@ function pruneJobs(now = Date.now()) {
|
||||
for (const [jobId, job] of jobs.entries()) {
|
||||
if (
|
||||
job.subscribers.size === 0 &&
|
||||
(job.status === "success" || job.status === "error") &&
|
||||
(job.status === "success" ||
|
||||
job.status === "error" ||
|
||||
job.status === "cancelled") &&
|
||||
now - job.updatedAt > JOB_TTL_MS
|
||||
) {
|
||||
jobs.delete(jobId);
|
||||
@@ -102,14 +112,39 @@ export function updateProcessJob(
|
||||
status?: ProcessJobStatus;
|
||||
result?: unknown;
|
||||
error?: string;
|
||||
cancel?: (reason?: string) => void;
|
||||
cancelReason?: string;
|
||||
},
|
||||
) {
|
||||
job.status = patch.status ?? job.status;
|
||||
job.result = patch.result ?? job.result;
|
||||
job.error = patch.error ?? job.error;
|
||||
job.cancel = patch.cancel ?? job.cancel;
|
||||
job.cancelReason = patch.cancelReason ?? job.cancelReason;
|
||||
job.updatedAt = Date.now();
|
||||
}
|
||||
|
||||
export function cancelProcessJob(job: ProcessJob, reason = "Cancelled by user") {
|
||||
if (job.status === "success" || job.status === "error") {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (job.status === "cancelled") {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (job.status === "cancelling") {
|
||||
return true;
|
||||
}
|
||||
|
||||
job.status = "cancelling";
|
||||
job.cancelReason = reason;
|
||||
job.updatedAt = Date.now();
|
||||
job.cancel?.(reason);
|
||||
appendProcessJobLog(job, `[job] Cancellation requested: ${reason}`);
|
||||
return true;
|
||||
}
|
||||
|
||||
export function subscribeProcessJob(
|
||||
job: ProcessJob,
|
||||
subscriber: (entry: ProcessJobLogEntry) => void,
|
||||
@@ -131,5 +166,6 @@ export function serializeProcessJob(job: ProcessJob) {
|
||||
logs: job.logs,
|
||||
result: job.result,
|
||||
error: job.error,
|
||||
cancelReason: job.cancelReason,
|
||||
};
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
import type { SSHExecCommandResponse } from "node-ssh";
|
||||
import { Server } from "@prisma/client";
|
||||
import { AsyncLocalStorage } from "async_hooks";
|
||||
import { randomUUID } from "crypto";
|
||||
|
||||
import {
|
||||
closeConnection,
|
||||
@@ -11,12 +12,86 @@ import {
|
||||
const commandQueues = new Map<string, Promise<unknown>>();
|
||||
const streamCommandQueues = new Map<string, Promise<unknown>>();
|
||||
const commandLogSink = new AsyncLocalStorage<(data: string) => void>();
|
||||
const commandRunContext = new AsyncLocalStorage<{
|
||||
runIdPrefix?: string;
|
||||
signal?: AbortSignal;
|
||||
}>();
|
||||
|
||||
type ExecOptions = {
|
||||
timeoutMs?: number;
|
||||
queueTimeoutMs?: number;
|
||||
};
|
||||
|
||||
const REMOTE_JOB_DIR = "/tmp/doktainer-process-jobs";
|
||||
|
||||
function shellQuote(value: string) {
|
||||
return `'${value.replace(/'/g, `'\\''`)}'`;
|
||||
}
|
||||
|
||||
function sanitizeRunId(value: string) {
|
||||
return value.replace(/[^a-zA-Z0-9_.-]/g, "-").slice(0, 120);
|
||||
}
|
||||
|
||||
function buildTrackedCommand(command: string, runId: string) {
|
||||
const safeRunId = sanitizeRunId(runId);
|
||||
const script = [
|
||||
"set -euo pipefail",
|
||||
`RUN_DIR=${shellQuote(REMOTE_JOB_DIR)}`,
|
||||
`RUN_ID=${shellQuote(safeRunId)}`,
|
||||
'mkdir -p "$RUN_DIR"',
|
||||
'PID_FILE="$RUN_DIR/$RUN_ID.pid"',
|
||||
'rm -f "$PID_FILE"',
|
||||
'cleanup() { rm -f "$PID_FILE"; }',
|
||||
"trap cleanup EXIT",
|
||||
`setsid bash -lc ${shellQuote(command)} &`,
|
||||
"child=$!",
|
||||
'printf "%s" "$child" > "$PID_FILE"',
|
||||
"set +e",
|
||||
'wait "$child"',
|
||||
"code=$?",
|
||||
"set -e",
|
||||
"exit \"$code\"",
|
||||
].join("\n");
|
||||
|
||||
return `bash -lc ${shellQuote(script)}`;
|
||||
}
|
||||
|
||||
async function terminateTrackedRemoteCommand(
|
||||
server: Server,
|
||||
runId: string,
|
||||
reason: string,
|
||||
) {
|
||||
const safeRunId = sanitizeRunId(runId);
|
||||
const script = [
|
||||
"set -euo pipefail",
|
||||
`PID_FILE=${shellQuote(`${REMOTE_JOB_DIR}/${safeRunId}.pid`)}`,
|
||||
'if [ ! -f "$PID_FILE" ]; then exit 0; fi',
|
||||
'pid="$(cat "$PID_FILE" 2>/dev/null || true)"',
|
||||
'case "$pid" in ""|*[!0-9]*) rm -f "$PID_FILE"; exit 0 ;; esac',
|
||||
`printf "Terminating remote command ${safeRunId}: ${reason.replace(/"/g, "'")}\\n" >&2`,
|
||||
'kill -TERM -- "-$pid" 2>/dev/null || kill -TERM "$pid" 2>/dev/null || sudo -n kill -TERM -- "-$pid" 2>/dev/null || sudo -n kill -TERM "$pid" 2>/dev/null || true',
|
||||
"sleep 5",
|
||||
'kill -KILL -- "-$pid" 2>/dev/null || kill -KILL "$pid" 2>/dev/null || sudo -n kill -KILL -- "-$pid" 2>/dev/null || sudo -n kill -KILL "$pid" 2>/dev/null || true',
|
||||
'rm -f "$PID_FILE"',
|
||||
].join("\n");
|
||||
|
||||
try {
|
||||
await withIsolatedConnection(server, async (ssh) => {
|
||||
await ssh.execCommand(`bash -lc ${shellQuote(script)}`);
|
||||
});
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
console.warn(
|
||||
`[ssh-cancel-failed] ${JSON.stringify({
|
||||
serverId: server.id,
|
||||
host: server.ip,
|
||||
runId: safeRunId,
|
||||
error: message,
|
||||
})}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function isRecoverableSshError(message: string): boolean {
|
||||
return (
|
||||
message.includes("Channel open failure") ||
|
||||
@@ -308,6 +383,10 @@ export async function execStrict(
|
||||
): Promise<string> {
|
||||
const logSink = commandLogSink.getStore();
|
||||
if (logSink) {
|
||||
const runContext = commandRunContext.getStore();
|
||||
const runId = runContext?.runIdPrefix
|
||||
? `${runContext.runIdPrefix}-${randomUUID()}`
|
||||
: undefined;
|
||||
let output = "";
|
||||
let exitCode = 0;
|
||||
|
||||
@@ -321,6 +400,11 @@ export async function execStrict(
|
||||
(code) => {
|
||||
exitCode = code;
|
||||
},
|
||||
{
|
||||
runId,
|
||||
signal: runContext?.signal,
|
||||
timeoutMs: options.timeoutMs,
|
||||
},
|
||||
);
|
||||
|
||||
if (exitCode !== 0) {
|
||||
@@ -344,8 +428,11 @@ export async function execStrict(
|
||||
export function withCommandLogSink<T>(
|
||||
onData: (data: string) => void,
|
||||
execute: () => Promise<T>,
|
||||
options: { runIdPrefix?: string; signal?: AbortSignal } = {},
|
||||
): Promise<T> {
|
||||
return commandLogSink.run(onData, execute);
|
||||
return commandLogSink.run(onData, () =>
|
||||
commandRunContext.run(options, execute),
|
||||
);
|
||||
}
|
||||
|
||||
export async function execStrictIsolated(
|
||||
@@ -371,19 +458,48 @@ export async function streamCommand(
|
||||
command: string,
|
||||
onData: (data: string) => void,
|
||||
onClose: (code: number) => void,
|
||||
options: { timeoutMs?: number; signal?: AbortSignal; runId?: string } = {},
|
||||
): Promise<void> {
|
||||
return enqueueStreamCommand(server, command, async () => {
|
||||
const ssh = await getConnection(server);
|
||||
const commandToRun = options.runId
|
||||
? buildTrackedCommand(command, options.runId)
|
||||
: command;
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
ssh.connection!.exec(command, (err, stream) => {
|
||||
let streamRef: { destroy: () => void } | null = null;
|
||||
let timeoutId: NodeJS.Timeout | null = null;
|
||||
let abortListener: ((event: Event) => void) | null = null;
|
||||
|
||||
const terminate = (reason: string) => {
|
||||
if (options.runId) {
|
||||
void terminateTrackedRemoteCommand(server, options.runId, reason);
|
||||
}
|
||||
|
||||
try {
|
||||
streamRef?.destroy();
|
||||
} catch {
|
||||
// Ignore stream races during remote cancellation.
|
||||
}
|
||||
|
||||
closeConnection(server.id);
|
||||
};
|
||||
|
||||
ssh.connection!.exec(commandToRun, (err, stream) => {
|
||||
if (err) {
|
||||
reject(err);
|
||||
return;
|
||||
}
|
||||
|
||||
streamRef = stream;
|
||||
let settled = false;
|
||||
const cleanup = () => {
|
||||
if (timeoutId) {
|
||||
clearTimeout(timeoutId);
|
||||
}
|
||||
if (abortListener && options.signal) {
|
||||
options.signal.removeEventListener("abort", abortListener);
|
||||
}
|
||||
stream.off("data", onStdout);
|
||||
stream.stderr?.off("data", onStderr);
|
||||
stream.off("close", onCloseStream);
|
||||
@@ -417,6 +533,38 @@ export async function streamCommand(
|
||||
stream.on("end", onEnd);
|
||||
stream.on("error", onError);
|
||||
stream.stderr?.on("error", onError);
|
||||
|
||||
if (options.timeoutMs && options.timeoutMs > 0) {
|
||||
timeoutId = setTimeout(() => {
|
||||
if (settled) return;
|
||||
logCommandTimeout(server, command, options.timeoutMs!);
|
||||
terminate(
|
||||
`Command timed out after ${Math.ceil(options.timeoutMs! / 1000)}s`,
|
||||
);
|
||||
fail(
|
||||
new Error(
|
||||
`Command timed out after ${Math.ceil(options.timeoutMs! / 1000)}s`,
|
||||
),
|
||||
);
|
||||
}, options.timeoutMs);
|
||||
timeoutId.unref?.();
|
||||
}
|
||||
|
||||
if (options.signal) {
|
||||
if (options.signal.aborted) {
|
||||
terminate("Command cancelled");
|
||||
fail(new Error("Command cancelled"));
|
||||
return;
|
||||
}
|
||||
|
||||
abortListener = () => {
|
||||
terminate("Command cancelled");
|
||||
fail(new Error("Command cancelled"));
|
||||
};
|
||||
options.signal.addEventListener("abort", abortListener, {
|
||||
once: true,
|
||||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -22,6 +22,9 @@ const DOCKER_FILE_LIST_TIMEOUT_MS = 10_000;
|
||||
const DOCKER_FILE_READ_TIMEOUT_MS = 15_000;
|
||||
const DOCKER_FILE_WRITE_TIMEOUT_MS = 20_000;
|
||||
const DOCKER_FILE_DOWNLOAD_TIMEOUT_MS = 30_000;
|
||||
const DEPLOY_GIT_CLONE_TIMEOUT_MS = 10 * 60_000;
|
||||
const DEPLOY_BUILD_TIMEOUT_MS = 45 * 60_000;
|
||||
const DEPLOY_COMPOSE_TIMEOUT_MS = 45 * 60_000;
|
||||
|
||||
function shortDockerCommandTimeout(timeoutMs: number) {
|
||||
return { timeoutMs, queueTimeoutMs: timeoutMs };
|
||||
@@ -1953,6 +1956,7 @@ async function buildImageWithNixpacks(
|
||||
await execStrict(
|
||||
server,
|
||||
privilegedCommand(server, `bash -lc ${escapeShellArg(script)}`),
|
||||
shortDockerCommandTimeout(DEPLOY_BUILD_TIMEOUT_MS),
|
||||
);
|
||||
|
||||
await ensureImageExists(server, opts.imageTag, "Nixpacks build");
|
||||
@@ -2005,6 +2009,7 @@ async function buildImageWithBuildpacks(
|
||||
await execStrict(
|
||||
server,
|
||||
privilegedCommand(server, `bash -lc ${escapeShellArg(script)}`),
|
||||
shortDockerCommandTimeout(DEPLOY_BUILD_TIMEOUT_MS),
|
||||
);
|
||||
|
||||
await ensureImageExists(server, opts.imageTag, "Buildpacks build");
|
||||
@@ -2395,6 +2400,7 @@ export async function deployContainerFromGitSource(
|
||||
await execStrict(
|
||||
server,
|
||||
privilegedCommand(server, `bash -lc ${escapeShellArg(bootstrapScript)}`),
|
||||
shortDockerCommandTimeout(DEPLOY_GIT_CLONE_TIMEOUT_MS),
|
||||
);
|
||||
} catch (error) {
|
||||
throw new Error(formatDeploymentErrorMessage(error));
|
||||
@@ -2427,6 +2433,7 @@ export async function deployContainerFromGitSource(
|
||||
await execStrict(
|
||||
server,
|
||||
privilegedCommand(server, `bash -lc ${escapeShellArg(composeScript)}`),
|
||||
shortDockerCommandTimeout(DEPLOY_COMPOSE_TIMEOUT_MS),
|
||||
);
|
||||
} catch (error) {
|
||||
throw new Error(formatComposeDeployError(error));
|
||||
@@ -2693,6 +2700,7 @@ export async function deployContainerFromGitSource(
|
||||
await execStrict(
|
||||
server,
|
||||
privilegedCommand(server, `bash -lc ${escapeShellArg(buildScript)}`),
|
||||
shortDockerCommandTimeout(DEPLOY_BUILD_TIMEOUT_MS),
|
||||
);
|
||||
} catch (error) {
|
||||
throw new Error(formatDeploymentErrorMessage(error));
|
||||
|
||||
Reference in new issue
Block a user