378 lines
12 KiB
TypeScript
378 lines
12 KiB
TypeScript
import { spawn, ChildProcess } from "child_process";
|
|
import fs from "fs";
|
|
import path from "path";
|
|
import {
|
|
DATA_DIR,
|
|
Pipeline,
|
|
Run,
|
|
createRun,
|
|
db,
|
|
getConnection,
|
|
getPipeline,
|
|
getRun,
|
|
updateRunStatus,
|
|
} from "./db";
|
|
import { SLING_HOME, buildReplicationYaml, connectionUrl } from "./sling";
|
|
import { buildApplyArgs, computeSchemaExcludes, prepareDevDatabase } from "./atlas";
|
|
import { resolveBinary } from "./binaries";
|
|
|
|
export interface JobEvent {
|
|
line: string;
|
|
ts: number;
|
|
}
|
|
|
|
export type LineListener = (ev: JobEvent) => void;
|
|
export type DoneListener = (status: string) => void;
|
|
|
|
// Mask credentials in echoed command lines (e.g. mysql://user:pass@host/db)
|
|
function maskSecret(arg: string): string {
|
|
return arg.replace(/(\w[\w-]*:\/\/[^:/\s]+:)[^@\s]+(@)/, "$1****$2");
|
|
}
|
|
|
|
export class AlreadyRunningError extends Error {
|
|
constructor(pipelineId: number) {
|
|
super(`pipeline ${pipelineId} already has a running run`);
|
|
this.name = "AlreadyRunningError";
|
|
}
|
|
}
|
|
|
|
interface ActiveRun {
|
|
runId: number;
|
|
pipelineId: number;
|
|
logPath: string;
|
|
buffer: JobEvent[];
|
|
tail: string[]; // last lines, used for the error summary
|
|
listeners: Set<LineListener>;
|
|
doneListeners: Set<DoneListener>;
|
|
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<number, ActiveRun>(); // runId -> state
|
|
private byPipeline = new Map<number, number>(); // pipelineId -> runId
|
|
|
|
isRunning(pipelineId: number): boolean {
|
|
return this.byPipeline.has(pipelineId);
|
|
}
|
|
|
|
startRun(pipelineId: number): Run {
|
|
const pipeline = getPipeline(pipelineId);
|
|
if (!pipeline) throw new Error(`pipeline ${pipelineId} not found`);
|
|
if (this.isRunning(pipelineId)) throw new AlreadyRunningError(pipelineId);
|
|
|
|
const run = createRun({
|
|
pipeline_id: pipelineId,
|
|
status: "running",
|
|
started_at: new Date().toISOString(),
|
|
});
|
|
const runsDir = path.join(DATA_DIR, "runs");
|
|
fs.mkdirSync(runsDir, { recursive: true });
|
|
const logPath = path.join(runsDir, `${run.id}.log`);
|
|
updateRunStatus(run.id, { log_path: logPath });
|
|
|
|
const state: ActiveRun = {
|
|
runId: run.id,
|
|
pipelineId,
|
|
logPath,
|
|
buffer: [],
|
|
tail: [],
|
|
listeners: new Set(),
|
|
doneListeners: new Set(),
|
|
child: null,
|
|
cancelled: false,
|
|
};
|
|
this.active.set(run.id, state);
|
|
this.byPipeline.set(pipelineId, run.id);
|
|
|
|
void this.execute(pipeline, state);
|
|
return getRun(run.id)!;
|
|
}
|
|
|
|
// Subscribe to live output. Replays the in-memory buffer first, then pushes
|
|
// new lines. Returns an unsubscribe function, or null if the run is not
|
|
// active in this process (e.g. after a dev HMR reload).
|
|
subscribe(
|
|
runId: number,
|
|
onLine: LineListener,
|
|
onDone?: DoneListener
|
|
): (() => void) | null {
|
|
const state = this.active.get(runId);
|
|
if (!state) return null;
|
|
for (const ev of state.buffer) onLine(ev);
|
|
state.listeners.add(onLine);
|
|
if (onDone) state.doneListeners.add(onDone);
|
|
return () => {
|
|
state.listeners.delete(onLine);
|
|
if (onDone) state.doneListeners.delete(onDone);
|
|
};
|
|
}
|
|
|
|
private emit(state: ActiveRun, line: string) {
|
|
const ev: JobEvent = { line, ts: Date.now() };
|
|
state.buffer.push(ev);
|
|
state.tail.push(line);
|
|
if (state.tail.length > TAIL_LIMIT) state.tail.shift();
|
|
try {
|
|
fs.appendFileSync(state.logPath, line + "\n");
|
|
} catch {
|
|
// keep streaming to subscribers even if the log file fails
|
|
}
|
|
for (const l of state.listeners) l(ev);
|
|
}
|
|
|
|
// 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,
|
|
finished_at: new Date().toISOString(),
|
|
});
|
|
this.active.delete(state.runId);
|
|
this.byPipeline.delete(state.pipelineId);
|
|
for (const cb of state.doneListeners) cb(status);
|
|
state.listeners.clear();
|
|
state.doneListeners.clear();
|
|
}
|
|
|
|
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);
|
|
if (!src || !tgt) throw new Error("pipeline connections not found");
|
|
const excludes = await computeSchemaExcludes(pipeline);
|
|
const scope = pipeline.schema_scope || "selected";
|
|
const tableExcludes = pipeline.include_fk
|
|
? excludes.length
|
|
: excludes.filter((p) => p !== "*[type=foreign-key]").length;
|
|
this.emit(
|
|
state,
|
|
`[schema] scope=${scope}, excluding ${tableExcludes} objects, fk=${pipeline.include_fk ? "on" : "off"}`
|
|
);
|
|
const devUrl = prepareDevDatabase(tgt);
|
|
const code = await this.runProcess(
|
|
state,
|
|
resolveBinary("atlas") ?? "atlas",
|
|
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")}`
|
|
);
|
|
}
|
|
}
|
|
|
|
if (state.cancelled) return finishCancelled();
|
|
const yamlPath = buildReplicationYaml(pipeline, state.runId);
|
|
const code = await this.runProcess(
|
|
state,
|
|
resolveBinary("sling") ?? "sling",
|
|
["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 {
|
|
this.finish(
|
|
state,
|
|
"failed",
|
|
`sling exited with code ${code}\n${state.tail.join("\n")}`
|
|
);
|
|
}
|
|
} catch (err) {
|
|
if (state.cancelled) return finishCancelled();
|
|
this.emit(state, `[error] ${(err as Error).message}`);
|
|
this.finish(state, "failed", (err as Error).message);
|
|
}
|
|
}
|
|
|
|
private runProcess(
|
|
state: ActiveRun,
|
|
bin: string,
|
|
args: string[],
|
|
extraEnv: Record<string, string>
|
|
): Promise<number> {
|
|
return new Promise((resolve, reject) => {
|
|
this.emit(state, `$ ${bin} ${args.map(maskSecret).join(" ")}`);
|
|
const child = spawn(bin, args, {
|
|
env: { ...process.env, ...extraEnv },
|
|
});
|
|
state.child = child;
|
|
if (child.pid) {
|
|
updateRunStatus(state.runId, { pid: child.pid });
|
|
}
|
|
child.on("error", reject);
|
|
|
|
const hookStream = (stream: NodeJS.ReadableStream | null) => {
|
|
if (!stream) return;
|
|
let buf = "";
|
|
stream.on("data", (chunk: Buffer) => {
|
|
buf += chunk.toString();
|
|
let idx: number;
|
|
while ((idx = buf.indexOf("\n")) >= 0) {
|
|
this.emit(state, buf.slice(0, idx));
|
|
buf = buf.slice(idx + 1);
|
|
}
|
|
});
|
|
stream.on("end", () => {
|
|
if (buf) this.emit(state, buf);
|
|
});
|
|
};
|
|
hookStream(child.stdout);
|
|
hookStream(child.stderr);
|
|
|
|
child.on("close", (code) => {
|
|
if (state.child === child) state.child = null;
|
|
resolve(code ?? 1);
|
|
});
|
|
});
|
|
}
|
|
}
|
|
|
|
// Singleton surviving dev HMR reloads.
|
|
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);
|
|
}
|
|
}
|