diff --git a/lib/db.ts b/lib/db.ts index 380b59d..4236467 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -82,6 +82,7 @@ export interface Run { log_path: string | null; error: string | null; pid: number | null; + owner_pid: number | null; } function createDb(): Database.Database { @@ -123,7 +124,8 @@ function createDb(): Database.Database { finished_at TEXT, log_path TEXT, error TEXT, - pid INTEGER + pid INTEGER, + owner_pid INTEGER ); `); // Lightweight migrations for databases created before these columns existed. @@ -162,6 +164,11 @@ function createDb(): Database.Database { "ALTER TABLE pipelines ADD COLUMN env TEXT NOT NULL DEFAULT '{}'", "env" ); + addColumnIfMissing( + "runs", + "ALTER TABLE runs ADD COLUMN owner_pid INTEGER", + "owner_pid" + ); return db; } @@ -294,11 +301,12 @@ export function createRun(input: { started_at?: string; log_path?: string; pid?: number; + owner_pid?: number; }): Run { const res = db .prepare( - `INSERT INTO runs (pipeline_id, status, started_at, log_path, pid) - VALUES (@pipeline_id, @status, @started_at, @log_path, @pid)` + `INSERT INTO runs (pipeline_id, status, started_at, log_path, pid, owner_pid) + VALUES (@pipeline_id, @status, @started_at, @log_path, @pid, @owner_pid)` ) .run({ pipeline_id: input.pipeline_id, @@ -306,6 +314,7 @@ export function createRun(input: { started_at: input.started_at ?? null, log_path: input.log_path ?? null, pid: input.pid ?? null, + owner_pid: input.owner_pid ?? null, }); return getRun(Number(res.lastInsertRowid))!; } diff --git a/lib/jobs.ts b/lib/jobs.ts index 0160d60..ae7ed8c 100644 --- a/lib/jobs.ts +++ b/lib/jobs.ts @@ -84,6 +84,7 @@ class JobManager { pipeline_id: pipelineId, status: "running", started_at: new Date().toISOString(), + owner_pid: process.pid, }); const runsDir = path.join(DATA_DIR, "runs"); fs.mkdirSync(runsDir, { recursive: true }); @@ -324,16 +325,21 @@ 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. +// Each run records owner_pid (the Node process that started it). A run whose +// owner is still alive belongs to a live worker of the current server — skip +// it, otherwise a newly loaded worker would kill healthy runs mid-flight. +// Only runs with a dead (or legacy null) owner are reconciled: dead child pid +// -> failed; live child pid -> orphaned sling/atlas subprocess, SIGTERM it +// and mark 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) { + if (run.owner_pid != null && isProcessAlive(run.owner_pid)) { + continue; + } const alive = run.pid != null && isProcessAlive(run.pid); if (alive) { try {