|
@@ -1,5 +1,5 @@
|
|
|
import { readFileSync, writeFileSync, existsSync, mkdirSync } from "node:fs";
|
|
import { readFileSync, writeFileSync, existsSync, mkdirSync } from "node:fs";
|
|
|
-import { join } from "node:path";
|
|
|
|
|
|
|
+import { join, resolve } from "node:path";
|
|
|
import { tmpdir } from "node:os";
|
|
import { tmpdir } from "node:os";
|
|
|
|
|
|
|
|
export interface JobData {
|
|
export interface JobData {
|
|
@@ -31,9 +31,33 @@ export interface JobData {
|
|
|
updatedAt: number;
|
|
updatedAt: number;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-const STORE_DIR = join(tmpdir(), "pipeline-jobs");
|
|
|
|
|
|
|
+// Mountable job store. In Docker Compose this is /tmp/pipeline-jobs (a named
|
|
|
|
|
+// volume); in Kubernetes set PIPELINE_JOBS_DIR to a PVC-backed path so jobs
|
|
|
|
|
+// survive pod restarts instead of living on ephemeral container storage.
|
|
|
|
|
+const STORE_DIR = process.env.PIPELINE_JOBS_DIR
|
|
|
|
|
+ ? resolve(process.env.PIPELINE_JOBS_DIR)
|
|
|
|
|
+ : join(tmpdir(), "pipeline-jobs");
|
|
|
const STORE_FILE = join(STORE_DIR, "jobs.json");
|
|
const STORE_FILE = join(STORE_DIR, "jobs.json");
|
|
|
|
|
|
|
|
|
|
+// Rendering runs as a spawned child process; if the pod is restarted (rolling
|
|
|
|
|
+// update, OOM, node drain) mid-job, the job is left "running" forever. On the
|
|
|
|
|
+// first store access after a fresh start, sweep any interrupted jobs to failed
|
|
|
|
|
+// so the UI reflects reality instead of freezing on a ghost task.
|
|
|
|
|
+let recovered = false;
|
|
|
|
|
+function sweepInterrupted(jobs: Map<string, JobData>): boolean {
|
|
|
|
|
+ let dirty = false;
|
|
|
|
|
+ const now = Date.now();
|
|
|
|
|
+ for (const job of jobs.values()) {
|
|
|
|
|
+ if (job.status === "running" || job.status === "pending") {
|
|
|
|
|
+ job.status = "failed";
|
|
|
|
|
+ job.error = job.error ?? "任务因进程重启被中断(interrupted by pod restart)";
|
|
|
|
|
+ job.updatedAt = now;
|
|
|
|
|
+ dirty = true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return dirty;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
function ensureStore(): Map<string, JobData> {
|
|
function ensureStore(): Map<string, JobData> {
|
|
|
if (!existsSync(STORE_DIR)) {
|
|
if (!existsSync(STORE_DIR)) {
|
|
|
mkdirSync(STORE_DIR, { recursive: true });
|
|
mkdirSync(STORE_DIR, { recursive: true });
|
|
@@ -41,13 +65,19 @@ function ensureStore(): Map<string, JobData> {
|
|
|
if (!existsSync(STORE_FILE)) {
|
|
if (!existsSync(STORE_FILE)) {
|
|
|
writeFileSync(STORE_FILE, "{}", "utf-8");
|
|
writeFileSync(STORE_FILE, "{}", "utf-8");
|
|
|
}
|
|
}
|
|
|
|
|
+ let jobs: Map<string, JobData>;
|
|
|
try {
|
|
try {
|
|
|
const raw = readFileSync(STORE_FILE, "utf-8");
|
|
const raw = readFileSync(STORE_FILE, "utf-8");
|
|
|
const obj = JSON.parse(raw) as Record<string, JobData>;
|
|
const obj = JSON.parse(raw) as Record<string, JobData>;
|
|
|
- return new Map(Object.entries(obj));
|
|
|
|
|
|
|
+ jobs = new Map(Object.entries(obj));
|
|
|
} catch {
|
|
} catch {
|
|
|
- return new Map();
|
|
|
|
|
|
|
+ jobs = new Map();
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!recovered) {
|
|
|
|
|
+ recovered = true;
|
|
|
|
|
+ if (sweepInterrupted(jobs)) flushStore(jobs);
|
|
|
}
|
|
}
|
|
|
|
|
+ return jobs;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
function flushStore(jobs: Map<string, JobData>): void {
|
|
function flushStore(jobs: Map<string, JobData>): void {
|