diff --git a/app/api/runs/[id]/stop/route.ts b/app/api/runs/[id]/stop/route.ts new file mode 100644 index 0000000..add9b29 --- /dev/null +++ b/app/api/runs/[id]/stop/route.ts @@ -0,0 +1,34 @@ +import { getRun } from "@/lib/db"; +import { jobManager } from "@/lib/jobs"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function POST( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const { id } = await params; + const run = getRun(Number(id)); + if (!run) { + return Response.json({ error: "run not found" }, { status: 404 }); + } + if (run.status !== "running" && run.status !== "queued") { + return Response.json( + { error: `run ${run.id} 已结束（${run.status}）` }, + { status: 409 } + ); + } + const result = jobManager.stopRun(run.id); + if (!result.ok) { + return Response.json( + { error: result.message ?? "停止失败" }, + { status: 409 } + ); + } + return Response.json(result.run ?? getRun(run.id)); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/pipelines/[id]/page.tsx b/app/pipelines/[id]/page.tsx index 2422ef0..77fd5e9 100644 --- a/app/pipelines/[id]/page.tsx +++ b/app/pipelines/[id]/page.tsx @@ -15,6 +15,7 @@ import { } from "@/components/api"; import { LogViewer } from "@/components/log-viewer"; import { StatusBadge } from "@/components/status-badge"; +import { StopRunButton } from "@/components/stop-run-button"; import { fmtTime, fmtDuration } from "@/components/format"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; @@ -282,6 +283,10 @@ export default function PipelineDetailPage() { #{latestRun.id} + {latestRun.status === "running" || + latestRun.status === "queued" ? ( + + ) : null} ) : null} diff --git a/app/runs/[id]/page.tsx b/app/runs/[id]/page.tsx index 44de464..4209ae5 100644 --- a/app/runs/[id]/page.tsx +++ b/app/runs/[id]/page.tsx @@ -10,6 +10,7 @@ import { } from "@/components/api"; import { LogViewer } from "@/components/log-viewer"; import { StatusBadge } from "@/components/status-badge"; +import { StopRunButton } from "@/components/stop-run-button"; import { fmtTime, fmtDuration } from "@/components/format"; import { Card, @@ -56,6 +57,9 @@ export default function RunDetailPage() {

运行 #{run.id} + {isActive ? ( + + ) : null}

diff --git a/components/status-badge.tsx b/components/status-badge.tsx index adf4f10..7348546 100644 --- a/components/status-badge.tsx +++ b/components/status-badge.tsx @@ -15,6 +15,11 @@ const CONFIG: Record = { label: "失败", className: "border-red-500/40 bg-red-500/10 text-red-600 dark:text-red-400", }, + cancelled: { + label: "已取消", + className: + "border-yellow-500/40 bg-yellow-500/10 text-yellow-600 dark:text-yellow-400", + }, queued: { label: "排队中", className: diff --git a/components/stop-run-button.tsx b/components/stop-run-button.tsx new file mode 100644 index 0000000..8e31375 --- /dev/null +++ b/components/stop-run-button.tsx @@ -0,0 +1,61 @@ +"use client"; + +import { useState } from "react"; +import { toast } from "sonner"; +import { Square } from "lucide-react"; +import { api, ApiError } from "@/components/api"; +import { ConfirmDialog } from "@/components/confirm-dialog"; +import { Button } from "@/components/ui/button"; + +interface StopRunButtonProps { + runId: number; + onStopped?: () => void; +} + +export function StopRunButton({ runId, onStopped }: StopRunButtonProps) { + const [open, setOpen] = useState(false); + const [loading, setLoading] = useState(false); + + async function stop() { + setLoading(true); + try { + await api(`/api/runs/${runId}/stop`, { method: "POST" }); + toast.success("已发送停止请求"); + setOpen(false); + onStopped?.(); + } catch (e) { + if (e instanceof ApiError && e.status === 409) { + toast.info("运行已结束"); + setOpen(false); + onStopped?.(); + } else { + toast.error(e instanceof Error ? e.message : "停止失败"); + } + } finally { + setLoading(false); + } + } + + return ( + <> + + + + ); +} diff --git a/instrumentation.ts b/instrumentation.ts new file mode 100644 index 0000000..2061cf4 --- /dev/null +++ b/instrumentation.ts @@ -0,0 +1,7 @@ +export async function register() { + if (process.env.NEXT_RUNTIME === "nodejs") { + // Importing lib/jobs runs its module init: reconcile runs left in + // running/queued by a previous server process (orphan cleanup). + await import("./lib/jobs"); + } +} diff --git a/lib/db.ts b/lib/db.ts index e04995f..380b59d 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -64,7 +64,7 @@ export interface PipelineInput { env?: string; } -export type RunStatus = "queued" | "running" | "success" | "failed"; +export type RunStatus = "queued" | "running" | "success" | "failed" | "cancelled"; // Normalize a JSON-object column value from an API body: accept either a // plain object or a pre-serialized JSON string. diff --git a/lib/jobs.ts b/lib/jobs.ts index 6037adf..0160d60 100644 --- a/lib/jobs.ts +++ b/lib/jobs.ts @@ -1,4 +1,4 @@ -import { spawn } from "child_process"; +import { spawn, ChildProcess } from "child_process"; import fs from "fs"; import path from "path"; import { @@ -6,6 +6,7 @@ import { Pipeline, Run, createRun, + db, getConnection, getPipeline, getRun, @@ -43,10 +44,29 @@ interface ActiveRun { tail: string[]; // last lines, used for the error summary listeners: Set; doneListeners: Set; + child: ChildProcess | null; // current subprocess (atlas or sling) + cancelled: boolean; } const TAIL_LIMIT = 20; +// True if a process with this pid exists (EPERM means it exists but is owned +// by another user). +export function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (e) { + return (e as NodeJS.ErrnoException).code === "EPERM"; + } +} + +export interface StopResult { + ok: boolean; + message?: string; + run?: Run | null; +} + class JobManager { private active = new Map(); // runId -> state private byPipeline = new Map(); // pipelineId -> runId @@ -78,6 +98,8 @@ class JobManager { tail: [], listeners: new Set(), doneListeners: new Set(), + child: null, + cancelled: false, }; this.active.set(run.id, state); this.byPipeline.set(pipelineId, run.id); @@ -118,7 +140,69 @@ class JobManager { for (const l of state.listeners) l(ev); } - private finish(state: ActiveRun, status: "success" | "failed", error: string | null) { + // Stop a run. If it is active in this process, flag it cancelled and + // SIGTERM the current subprocess (SIGKILL after 5s if it ignores SIGTERM); + // the execute loop observes `cancelled` and finishes as 'cancelled'. + // If the run is not active here but the DB still says running/queued, it + // belongs to a dead server process (HMR/restart): SIGTERM the recorded pid + // if still alive, otherwise just repair the stale DB row. + stopRun(runId: number): StopResult { + const state = this.active.get(runId); + if (state) { + state.cancelled = true; + const child = state.child; + if (child && child.pid && child.exitCode === null) { + this.emit(state, "[sling-ui] 收到停止请求，正在终止进程…"); + try { + child.kill("SIGTERM"); + } catch { + // already exiting + } + setTimeout(() => { + if (state.child === child && child.exitCode === null) { + try { + child.kill("SIGKILL"); + } catch { + // already gone + } + } + }, 5000).unref(); + } + return { ok: true, run: getRun(runId) }; + } + + const run = getRun(runId); + if (!run) return { ok: false, message: "run not found" }; + if (run.status !== "running" && run.status !== "queued") { + return { ok: false, message: `run ${runId} already ${run.status}`, run }; + } + const now = new Date().toISOString(); + if (run.pid != null && isProcessAlive(run.pid)) { + try { + process.kill(run.pid, "SIGTERM"); + } catch { + // died between the check and the kill + } + updateRunStatus(runId, { + status: "cancelled", + error: "用户手动停止", + finished_at: now, + }); + } else { + updateRunStatus(runId, { + status: "failed", + error: "进程已退出（服务重启或异常）", + finished_at: now, + }); + } + return { ok: true, run: getRun(runId) }; + } + + private finish( + state: ActiveRun, + status: "success" | "failed" | "cancelled", + error: string | null + ) { updateRunStatus(state.runId, { status, error, @@ -132,7 +216,12 @@ class JobManager { } private async execute(pipeline: Pipeline, state: ActiveRun) { + const finishCancelled = () => { + this.emit(state, "[sling-ui] 运行已被用户停止"); + this.finish(state, "cancelled", "用户手动停止"); + }; try { + if (state.cancelled) return finishCancelled(); if (pipeline.schema_sync) { const src = getConnection(pipeline.source_conn_id); const tgt = getConnection(pipeline.target_conn_id); @@ -153,6 +242,7 @@ class JobManager { buildApplyArgs(connectionUrl(src), connectionUrl(tgt), devUrl, excludes), {} ); + if (state.cancelled) return finishCancelled(); if (code !== 0) { throw new Error( `atlas schema apply exited with code ${code}\n${state.tail.join("\n")}` @@ -160,6 +250,7 @@ class JobManager { } } + if (state.cancelled) return finishCancelled(); const yamlPath = buildReplicationYaml(pipeline, state.runId); const code = await this.runProcess( state, @@ -167,6 +258,7 @@ class JobManager { ["run", "-r", yamlPath, "--home-dir", SLING_HOME], { SLING_LOGGING: "JSON", SLING_DISABLE_TELEMETRY: "1" } ); + if (state.cancelled) return finishCancelled(); if (code === 0) { this.finish(state, "success", null); } else { @@ -177,6 +269,7 @@ class JobManager { ); } } catch (err) { + if (state.cancelled) return finishCancelled(); this.emit(state, `[error] ${(err as Error).message}`); this.finish(state, "failed", (err as Error).message); } @@ -193,6 +286,7 @@ class JobManager { const child = spawn(bin, args, { env: { ...process.env, ...extraEnv }, }); + state.child = child; if (child.pid) { updateRunStatus(state.runId, { pid: child.pid }); } @@ -216,7 +310,10 @@ class JobManager { hookStream(child.stdout); hookStream(child.stderr); - child.on("close", (code) => resolve(code ?? 1)); + child.on("close", (code) => { + if (state.child === child) state.child = null; + resolve(code ?? 1); + }); }); } } @@ -225,3 +322,56 @@ class JobManager { const g = globalThis as unknown as { __jobManager?: JobManager }; export const jobManager: JobManager = g.__jobManager ?? (g.__jobManager = new JobManager()); + +// Reconcile runs left in running/queued by a previous server process. +// A run whose recorded pid is dead is stale (the subprocess died with the +// old server or never existed) -> failed. A run whose pid is still alive is +// an orphaned sling/atlas subprocess that kept running after the old server +// exited -> SIGTERM it and mark the run cancelled. +function reconcileOrphanRuns() { + const stale = db + .prepare("SELECT * FROM runs WHERE status IN ('running', 'queued')") + .all() as Run[]; + const now = new Date().toISOString(); + for (const run of stale) { + const alive = run.pid != null && isProcessAlive(run.pid); + if (alive) { + try { + process.kill(run.pid!, "SIGTERM"); + } catch { + // died between the check and the kill + } + updateRunStatus(run.id, { + status: "cancelled", + error: "服务重启，运行已终止", + finished_at: now, + }); + } else { + updateRunStatus(run.id, { + status: "failed", + error: "服务重启或进程异常退出", + finished_at: now, + }); + } + if (run.log_path && fs.existsSync(run.log_path)) { + try { + fs.appendFileSync(run.log_path, "[sling-ui] 服务重启，运行终止\n"); + } catch { + // log append is best-effort + } + } + console.log( + `[sling-ui] reconciled orphan run #${run.id} -> ${alive ? "cancelled" : "failed"} (pid ${run.pid ?? "none"})` + ); + } +} + +const gReconcile = globalThis as unknown as { __runsReconciled?: boolean }; +if (!gReconcile.__runsReconciled) { + gReconcile.__runsReconciled = true; + try { + reconcileOrphanRuns(); + } catch (e) { + console.error("[sling-ui] orphan run reconcile failed", e); + } +}