fix(web): make continuous handoff self-recovering
This commit is contained in:
@@ -16,10 +16,12 @@ const valueArg = (name, fallback) => {
|
||||
};
|
||||
const maxIterations = Number(valueArg("--max-iterations", "0"));
|
||||
const pollMs = Number(valueArg("--poll-ms", "1000"));
|
||||
const pushRetries = Number(valueArg("--push-retries", "6"));
|
||||
const commitEnabled = process.argv.includes("--commit");
|
||||
const pushEnabled = process.argv.includes("--push");
|
||||
if (!Number.isSafeInteger(maxIterations) || maxIterations < 0) throw new Error("--max-iterations must be a non-negative integer; 0 means unlimited");
|
||||
if (!Number.isSafeInteger(pollMs) || pollMs < 0) throw new Error("--poll-ms must be a non-negative integer");
|
||||
if (!Number.isSafeInteger(pushRetries) || pushRetries < 0) throw new Error("--push-retries must be a non-negative integer");
|
||||
if (pushEnabled && !commitEnabled) throw new Error("--push requires --commit");
|
||||
|
||||
fs.mkdirSync(abs(supervisorRoot), { recursive: true });
|
||||
@@ -59,19 +61,34 @@ const releaseLock = () => {
|
||||
try { fs.unlinkSync(abs(lockPath)); }
|
||||
catch (error) { if (error?.code !== "ENOENT") throw error; }
|
||||
};
|
||||
let stopRequested = false;
|
||||
let stopSignal;
|
||||
let iteration = 0;
|
||||
let fatalHandled = false;
|
||||
const stopFromSignal = (signal) => {
|
||||
try {
|
||||
writeCheckpoint({ state: "stopped", iteration: 0, reason: signal });
|
||||
appendLog(`stopped signal=${signal}`);
|
||||
releaseLock();
|
||||
}
|
||||
finally {
|
||||
process.exitCode = signal === "SIGINT" ? 130 : 143;
|
||||
}
|
||||
stopRequested = true;
|
||||
stopSignal = signal;
|
||||
appendLog(`stop-requested signal=${signal}`);
|
||||
process.exitCode = signal === "SIGINT" ? 130 : 143;
|
||||
};
|
||||
process.once("SIGINT", () => stopFromSignal("SIGINT"));
|
||||
process.once("SIGTERM", () => stopFromSignal("SIGTERM"));
|
||||
|
||||
const stopOnFatal = (error, source) => {
|
||||
if (fatalHandled) return;
|
||||
fatalHandled = true;
|
||||
try {
|
||||
writeCheckpoint({ state: "blocked", iteration, reason: source, error: error?.message ?? String(error) });
|
||||
appendLog(`blocked reason=${source} error=${error?.message ?? String(error)}`);
|
||||
}
|
||||
finally {
|
||||
releaseLock();
|
||||
process.exit(1);
|
||||
}
|
||||
};
|
||||
process.once("uncaughtException", (error) => stopOnFatal(error, "uncaught_exception"));
|
||||
process.once("unhandledRejection", (error) => stopOnFatal(error, "unhandled_rejection"));
|
||||
|
||||
if (lockHandle === undefined) {
|
||||
throw new Error(`unable to acquire supervisor lock=${lockPath}`);
|
||||
}
|
||||
@@ -96,14 +113,18 @@ const currentTask = () => {
|
||||
const childArgs = ["tools/web/run-wbv2-continuous.mjs", "--max-iterations", "1"];
|
||||
if (commitEnabled) childArgs.push("--commit");
|
||||
if (pushEnabled) childArgs.push("--push");
|
||||
if (pushEnabled) childArgs.push("--push-retries", String(pushRetries));
|
||||
|
||||
appendLog(`started pid=${process.pid} maxIterations=${maxIterations} child=${childArgs.join(" ")}`);
|
||||
writeCheckpoint({ state: "started", iteration: 0, current: currentTask(), child: childArgs.join(" ") });
|
||||
|
||||
let iteration = 0;
|
||||
let exitCode = 0;
|
||||
try {
|
||||
while (maxIterations === 0 || iteration < maxIterations) {
|
||||
if (stopRequested) {
|
||||
writeCheckpoint({ state: "stopped", iteration, current: currentTask(), reason: stopSignal });
|
||||
break;
|
||||
}
|
||||
const before = currentTask();
|
||||
const print = run(process.execPath, ["tools/web/print-task-context.mjs"]);
|
||||
const check = run(process.execPath, ["tools/web/check-task-context.mjs"]);
|
||||
@@ -117,6 +138,10 @@ try {
|
||||
const child = run(process.execPath, childArgs);
|
||||
appendLog(`child task=${before.task} exit=${child.code}\n${child.output}`);
|
||||
const after = currentTask();
|
||||
if (stopRequested) {
|
||||
writeCheckpoint({ state: "stopped", iteration, before, after, reason: stopSignal, childExit: child.code });
|
||||
break;
|
||||
}
|
||||
if (child.code !== 0) {
|
||||
const reason = after.task === before.task ? "child_failed_same_task" : "child_failed_after_queue_advance";
|
||||
writeCheckpoint({ state: "blocked", iteration: iteration + 1, before, after, reason, childExit: child.code });
|
||||
@@ -136,6 +161,9 @@ try {
|
||||
Atomics.wait(wait, 0, 0, pollMs);
|
||||
}
|
||||
}
|
||||
if (stopRequested) {
|
||||
exitCode = stopSignal === "SIGINT" ? 130 : 143;
|
||||
}
|
||||
if (exitCode === 0) {
|
||||
const current = currentTask();
|
||||
writeCheckpoint({ state: maxIterations === 0 ? "running" : "complete", iteration, current });
|
||||
|
||||
Reference in New Issue
Block a user