mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-25 11:55:47 -06:00
fix(cloud-workers): preserve accepted workspace after SSH loss (#120717)
* fix(cloud-workers): serialize accepted workspace rollback Prevent rollback and recovery from racing a remote accepted-workspace apply after SSH loses its exit status. Settle the same transaction nonce under a durable phase lock before deciding whether publication succeeded. Refs #120655 * refactor(gateway): keep workspace apply classifier private * refactor(gateway): split accepted workspace lock script * refactor(gateway): isolate accepted workspace lock Keep the byte-identical remote lock fragment with the accepted-workspace transaction instead of the manifest implementation. * test(gateway): fix settle manifest timing assertion * fix(cloud-workers): preserve indeterminate workspace publication Keep remote and local rollback journals pending when accepted-workspace settlement cannot be observed. Recovery now owns restoring both sides, while definitive apply and commit failures still roll back immediately. Refs #120655
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
23f36d27ee
commit
8fbb729ad3
@@ -0,0 +1,253 @@
|
||||
export const REMOTE_WORKSPACE_ACCEPTED_LOCK_JS = String.raw`const lockRoot = path.join(
|
||||
transactionRoot,
|
||||
".openclaw-accepted-lock-" + workspaceKey,
|
||||
);
|
||||
const lockToken = crypto.randomBytes(16).toString("hex");
|
||||
const lockOwner = { action, nonce, pid: process.pid, token: lockToken };
|
||||
const lockWait = new Int32Array(new SharedArrayBuffer(4));
|
||||
const lockDeadlineMs = Date.now() + 9 * 60 * 1000;
|
||||
let acquiredLock;
|
||||
function encodeLockIdentity(identity) {
|
||||
return [identity.action, identity.nonce, identity.pid, identity.token].join(".");
|
||||
}
|
||||
function parseLockIdentity(parts) {
|
||||
if (parts.length !== 4) return null;
|
||||
const [entryAction, entryNonce, rawPid, token] = parts;
|
||||
const pid = Number(rawPid);
|
||||
if (
|
||||
!acceptedActions.includes(entryAction) ||
|
||||
!/^[a-f0-9]{32}$/.test(entryNonce || "") ||
|
||||
!/^[1-9][0-9]*$/.test(rawPid || "") ||
|
||||
!Number.isSafeInteger(pid) ||
|
||||
!/^[a-f0-9]{32}$/.test(token || "")
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return { action: entryAction, nonce: entryNonce, pid, token };
|
||||
}
|
||||
function sameLockIdentity(left, right) {
|
||||
return (
|
||||
left.action === right.action &&
|
||||
left.nonce === right.nonce &&
|
||||
left.pid === right.pid &&
|
||||
left.token === right.token
|
||||
);
|
||||
}
|
||||
function ownerEntryName(owner) {
|
||||
return "owner." + encodeLockIdentity(owner);
|
||||
}
|
||||
function reclaimEntryName(owner, reclaimer) {
|
||||
return "reclaim." + encodeLockIdentity(owner) + "." + encodeLockIdentity(reclaimer);
|
||||
}
|
||||
function parseLockEntry(name) {
|
||||
const parts = name.split(".");
|
||||
if (parts[0] === "owner" && parts.length === 5) {
|
||||
const owner = parseLockIdentity(parts.slice(1));
|
||||
return owner ? { kind: "owner", owner } : null;
|
||||
}
|
||||
if (parts[0] === "reclaim" && parts.length === 9) {
|
||||
const owner = parseLockIdentity(parts.slice(1, 5));
|
||||
const reclaimer = parseLockIdentity(parts.slice(5));
|
||||
return owner && reclaimer ? { kind: "reclaim", owner, reclaimer } : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
function readLock() {
|
||||
let directoryStats;
|
||||
let names;
|
||||
try {
|
||||
directoryStats = fs.lstatSync(lockRoot);
|
||||
names = fs.readdirSync(lockRoot);
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return null;
|
||||
throw error;
|
||||
}
|
||||
if (directoryStats.isSymbolicLink() || !directoryStats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace transaction lock");
|
||||
}
|
||||
if (names.length !== 1) throw new Error("invalid accepted workspace transaction lock");
|
||||
const entry = parseLockEntry(names[0]);
|
||||
if (!entry) throw new Error("invalid accepted workspace transaction lock owner");
|
||||
const entryPath = path.join(lockRoot, names[0]);
|
||||
try {
|
||||
const entryStats = fs.lstatSync(entryPath);
|
||||
if (entryStats.isSymbolicLink() || !entryStats.isFile()) {
|
||||
throw new Error("unsafe accepted workspace transaction lock owner");
|
||||
}
|
||||
return { ...entry, name: names[0], entryPath, directoryStats, entryStats };
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
function sameLock(left, right) {
|
||||
return (
|
||||
left.kind === right.kind &&
|
||||
left.name === right.name &&
|
||||
sameInode(left.directoryStats, right.directoryStats) &&
|
||||
sameInode(left.entryStats, right.entryStats)
|
||||
);
|
||||
}
|
||||
function observedLockIdentity(lock) {
|
||||
return [
|
||||
lock.directoryStats.dev,
|
||||
lock.directoryStats.ino,
|
||||
lock.entryStats.dev,
|
||||
lock.entryStats.ino,
|
||||
lock.name,
|
||||
].join(":");
|
||||
}
|
||||
function restoreOwnerEntry(observed) {
|
||||
const current = readLock();
|
||||
if (!current || !sameLock(current, observed)) return false;
|
||||
try {
|
||||
fs.renameSync(current.entryPath, path.join(lockRoot, ownerEntryName(current.owner)));
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return false;
|
||||
throw error;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
function restoreAbandonedTransition(observed) {
|
||||
if (observed.kind !== "reclaim" || processIsAlive(observed.reclaimer.pid)) return false;
|
||||
const current = readLock();
|
||||
if (!current || !sameLock(current, observed) || processIsAlive(current.reclaimer.pid)) {
|
||||
return false;
|
||||
}
|
||||
return restoreOwnerEntry(current);
|
||||
}
|
||||
function reclaimDeadOwner(observed) {
|
||||
const current = readLock();
|
||||
if (
|
||||
!current ||
|
||||
current.kind !== "owner" ||
|
||||
!sameLock(current, observed) ||
|
||||
processIsAlive(current.owner.pid)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
const claimName = reclaimEntryName(current.owner, lockOwner);
|
||||
try {
|
||||
// This sole-entry rename is the reclaim CAS. Only one dead-owner contender
|
||||
// can install its complete owner+reclaimer identity.
|
||||
fs.renameSync(current.entryPath, path.join(lockRoot, claimName));
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return false;
|
||||
throw error;
|
||||
}
|
||||
let claimed = readLock();
|
||||
if (
|
||||
!claimed ||
|
||||
claimed.kind !== "reclaim" ||
|
||||
claimed.name !== claimName ||
|
||||
!sameInode(claimed.directoryStats, current.directoryStats) ||
|
||||
!sameInode(claimed.entryStats, current.entryStats) ||
|
||||
!sameLockIdentity(claimed.owner, current.owner) ||
|
||||
!sameLockIdentity(claimed.reclaimer, lockOwner)
|
||||
) {
|
||||
throw new Error("accepted workspace transaction reclaim ownership changed");
|
||||
}
|
||||
let quarantined = false;
|
||||
const quarantine = lockRoot + ".stale." + process.pid + "." + lockToken;
|
||||
try {
|
||||
if (processIsAlive(claimed.owner.pid)) return false;
|
||||
const validated = readLock();
|
||||
if (!validated || !sameLock(validated, claimed) || processIsAlive(validated.owner.pid)) {
|
||||
return false;
|
||||
}
|
||||
claimed = validated;
|
||||
fs.renameSync(lockRoot, quarantine);
|
||||
quarantined = true;
|
||||
const quarantinedDirectory = fs.lstatSync(quarantine);
|
||||
const quarantinedEntry = fs.lstatSync(path.join(quarantine, claimed.name));
|
||||
if (
|
||||
!sameInode(quarantinedDirectory, claimed.directoryStats) ||
|
||||
!sameInode(quarantinedEntry, claimed.entryStats)
|
||||
) {
|
||||
throw new Error("accepted workspace transaction lock changed during reclamation");
|
||||
}
|
||||
removeTree(quarantine);
|
||||
return true;
|
||||
} finally {
|
||||
if (!quarantined) restoreOwnerEntry(claimed);
|
||||
}
|
||||
}
|
||||
function acquireWorkspaceLock() {
|
||||
const candidate = lockRoot + "." + process.pid + "." + lockToken;
|
||||
const ownerName = ownerEntryName(lockOwner);
|
||||
fs.mkdirSync(candidate, { mode: 0o700 });
|
||||
fs.writeFileSync(path.join(candidate, ownerName), "", { flag: "wx", mode: 0o600 });
|
||||
let acquired = false;
|
||||
let previousIdentity = "";
|
||||
let waitMs = 10;
|
||||
try {
|
||||
while (Date.now() < lockDeadlineMs) {
|
||||
try {
|
||||
// The owner entry is complete before this atomic namespace operation,
|
||||
// so contenders never mistake an initializing live owner for stale.
|
||||
fs.renameSync(candidate, lockRoot);
|
||||
acquired = true;
|
||||
const observed = readLock();
|
||||
if (
|
||||
!observed ||
|
||||
observed.kind !== "owner" ||
|
||||
!sameLockIdentity(observed.owner, lockOwner)
|
||||
) {
|
||||
throw new Error("accepted workspace transaction lock acquisition changed");
|
||||
}
|
||||
acquiredLock = observed;
|
||||
return;
|
||||
} catch (error) {
|
||||
if (!error || (error.code !== "EEXIST" && error.code !== "ENOTEMPTY")) throw error;
|
||||
}
|
||||
const observed = readLock();
|
||||
if (!observed) {
|
||||
previousIdentity = "";
|
||||
waitMs = 10;
|
||||
continue;
|
||||
}
|
||||
const identity = observedLockIdentity(observed);
|
||||
if (identity !== previousIdentity) {
|
||||
previousIdentity = identity;
|
||||
waitMs = 10;
|
||||
}
|
||||
if (observed.kind !== "owner") {
|
||||
if (restoreAbandonedTransition(observed)) continue;
|
||||
} else if (!processIsAlive(observed.owner.pid) && reclaimDeadOwner(observed)) {
|
||||
continue;
|
||||
}
|
||||
Atomics.wait(lockWait, 0, 0, waitMs);
|
||||
waitMs = Math.min(waitMs * 2, 500);
|
||||
}
|
||||
throw new Error("timed out waiting for accepted workspace transaction lock");
|
||||
} finally {
|
||||
if (!acquired) removeTree(candidate);
|
||||
}
|
||||
}
|
||||
function releaseWorkspaceLock() {
|
||||
const current = readLock();
|
||||
if (
|
||||
!current ||
|
||||
!acquiredLock ||
|
||||
current.kind !== "owner" ||
|
||||
!sameLock(current, acquiredLock) ||
|
||||
!sameLockIdentity(current.owner, lockOwner)
|
||||
) {
|
||||
throw new Error("accepted workspace transaction lock ownership changed");
|
||||
}
|
||||
const validated = readLock();
|
||||
if (!validated || !sameLock(validated, current)) {
|
||||
throw new Error("accepted workspace transaction lock changed during release");
|
||||
}
|
||||
const quarantine = lockRoot + ".released." + process.pid + "." + lockToken;
|
||||
fs.renameSync(lockRoot, quarantine);
|
||||
const quarantinedDirectory = fs.lstatSync(quarantine);
|
||||
const quarantinedEntry = fs.lstatSync(path.join(quarantine, validated.name));
|
||||
if (
|
||||
!sameInode(quarantinedDirectory, validated.directoryStats) ||
|
||||
!sameInode(quarantinedEntry, validated.entryStats)
|
||||
) {
|
||||
throw new Error("accepted workspace transaction lock changed during release");
|
||||
}
|
||||
removeTree(quarantine);
|
||||
}`;
|
||||
@@ -0,0 +1,54 @@
|
||||
import { isRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
|
||||
const SETTLEMENT_OUTCOMES = new Set(["begun", "rolled-back", "applied", "committed"]);
|
||||
|
||||
export type AcceptedWorkspaceSettlementOutcome = "begun" | "rolled-back" | "applied" | "committed";
|
||||
|
||||
export class AcceptedWorkspacePublicationIndeterminateError extends Error {
|
||||
override name = "AcceptedWorkspacePublicationIndeterminateError";
|
||||
readonly observationFailure!: unknown;
|
||||
|
||||
constructor(
|
||||
readonly operation: "apply" | "commit",
|
||||
publicationFailure: unknown,
|
||||
observationFailure: unknown,
|
||||
) {
|
||||
super("Accepted workspace publication is indeterminate and requires recovery", {
|
||||
cause: publicationFailure,
|
||||
});
|
||||
Object.defineProperty(this, "observationFailure", { value: observationFailure });
|
||||
}
|
||||
}
|
||||
|
||||
export function isAcceptedWorkspacePublicationIndeterminateError(
|
||||
error: unknown,
|
||||
): error is AcceptedWorkspacePublicationIndeterminateError {
|
||||
return error instanceof AcceptedWorkspacePublicationIndeterminateError;
|
||||
}
|
||||
|
||||
export function parseAcceptedWorkspaceSettlement(
|
||||
stdout: string,
|
||||
): AcceptedWorkspaceSettlementOutcome {
|
||||
const lines = stdout.split(/\r?\n/u).filter(Boolean);
|
||||
if (lines.length !== 1) {
|
||||
throw new Error("Worker returned an invalid accepted workspace settlement outcome");
|
||||
}
|
||||
let value: unknown;
|
||||
try {
|
||||
value = JSON.parse(lines[0]!);
|
||||
} catch (error) {
|
||||
throw new Error("Worker returned an invalid accepted workspace settlement outcome", {
|
||||
cause: error,
|
||||
});
|
||||
}
|
||||
if (
|
||||
!isRecord(value) ||
|
||||
Object.keys(value).length !== 2 ||
|
||||
value.version !== 1 ||
|
||||
typeof value.outcome !== "string" ||
|
||||
!SETTLEMENT_OUTCOMES.has(value.outcome)
|
||||
) {
|
||||
throw new Error("Worker returned an invalid accepted workspace settlement outcome");
|
||||
}
|
||||
return value.outcome as AcceptedWorkspaceSettlementOutcome;
|
||||
}
|
||||
@@ -0,0 +1,505 @@
|
||||
import { REMOTE_WORKSPACE_ACCEPTED_LOCK_JS } from "./workspace-accepted-lock-remote-script.js";
|
||||
|
||||
export const REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS = String.raw`const crypto = require("node:crypto");
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const action = process.argv[1];
|
||||
const acceptedActions = ["begin", "apply", "rollback", "recover", "commit", "settle"];
|
||||
if (!acceptedActions.includes(action)) throw new Error("invalid accepted workspace transaction action");
|
||||
const root = fs.realpathSync(process.argv[2]);
|
||||
const nonce = process.argv[3];
|
||||
if (!/^[a-f0-9]{32}$/.test(nonce || "")) throw new Error("invalid accepted workspace transaction");
|
||||
// REMOTE_WORKSPACE_SETUP_SCRIPT creates and chmods every workspace parent for this worker.
|
||||
// Keeping the transaction beside the workspace makes all live swaps same-filesystem renames.
|
||||
const transactionRoot = path.dirname(root);
|
||||
const transactionRootStats = fs.lstatSync(transactionRoot);
|
||||
if (transactionRootStats.isSymbolicLink() || !transactionRootStats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace transaction directory");
|
||||
}
|
||||
const workspaceKey = crypto.createHash("sha256").update(root).digest("hex");
|
||||
const transactionPrefix = ".openclaw-accepted-" + workspaceKey + "-";
|
||||
const cleanupPrefix = ".openclaw-accepted-cleanup-" + workspaceKey + "-";
|
||||
const transaction = path.join(transactionRoot, transactionPrefix + nonce);
|
||||
const cleanup = path.join(transactionRoot, cleanupPrefix + nonce);
|
||||
const nextRoot = path.join(transaction, "next");
|
||||
const backupRoot = path.join(transaction, "backup");
|
||||
const pathsFile = path.join(transaction, "paths.json");
|
||||
const stateFile = path.join(transaction, "state.json");
|
||||
const ancestorModesFile = path.join(transaction, "ancestor-modes.json");
|
||||
function isSafeRelativePath(relative) {
|
||||
return (
|
||||
typeof relative === "string" &&
|
||||
relative &&
|
||||
!relative.includes("\\") &&
|
||||
!path.posix.isAbsolute(relative) &&
|
||||
path.posix.normalize(relative) === relative &&
|
||||
relative !== "." &&
|
||||
relative !== ".." &&
|
||||
relative !== ".git" &&
|
||||
!relative.startsWith(".git/") &&
|
||||
!relative.startsWith("../")
|
||||
);
|
||||
}
|
||||
function parsePaths(raw) {
|
||||
const values = JSON.parse(raw);
|
||||
if (!Array.isArray(values) || values.length > 25_000) {
|
||||
throw new Error("invalid accepted workspace paths");
|
||||
}
|
||||
const paths = [...new Set(values)];
|
||||
for (const relative of paths) {
|
||||
if (!isSafeRelativePath(relative)) throw new Error("unsafe accepted workspace path");
|
||||
}
|
||||
const selected = new Set(paths);
|
||||
// Directory modes are canonical, so a changed directory is added, removed, or
|
||||
// replaced and all of its accepted descendants are changed and staged too.
|
||||
return paths
|
||||
.filter((relative) => {
|
||||
const segments = relative.split("/");
|
||||
for (let index = 1; index < segments.length; index += 1) {
|
||||
if (selected.has(segments.slice(0, index).join("/"))) return false;
|
||||
}
|
||||
return true;
|
||||
})
|
||||
.sort();
|
||||
}
|
||||
function targetPath(base, relative) {
|
||||
return path.join(base, relative);
|
||||
}
|
||||
function livePath(relative) {
|
||||
const segments = relative.split("/");
|
||||
let parent = root;
|
||||
for (const segment of segments.slice(0, -1)) {
|
||||
parent = path.join(parent, segment);
|
||||
const stats = fs.lstatSync(parent);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
}
|
||||
return path.join(root, relative);
|
||||
}
|
||||
function exists(target) {
|
||||
try {
|
||||
fs.lstatSync(target);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return false;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
function removeTree(target) {
|
||||
let stats;
|
||||
try {
|
||||
stats = fs.lstatSync(target);
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return;
|
||||
throw error;
|
||||
}
|
||||
if (stats.isDirectory() && !stats.isSymbolicLink()) {
|
||||
fs.chmodSync(target, 0o700);
|
||||
for (const name of fs.readdirSync(target)) removeTree(path.join(target, name));
|
||||
fs.rmdirSync(target);
|
||||
} else {
|
||||
fs.unlinkSync(target);
|
||||
}
|
||||
}
|
||||
function sameInode(left, right) {
|
||||
return left.dev === right.dev && left.ino === right.ino;
|
||||
}
|
||||
function processIsAlive(pid) {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error && error.code === "EPERM") return true;
|
||||
if (error && error.code === "ESRCH") return false;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
${REMOTE_WORKSPACE_ACCEPTED_LOCK_JS}
|
||||
function readPaths() {
|
||||
return parsePaths(fs.readFileSync(pathsFile, "utf8"));
|
||||
}
|
||||
function readPhase(candidate, required = true) {
|
||||
let value;
|
||||
try {
|
||||
value = JSON.parse(fs.readFileSync(path.join(candidate, "phase.json"), "utf8"));
|
||||
} catch (error) {
|
||||
if (!required && error && error.code === "ENOENT") return null;
|
||||
throw error;
|
||||
}
|
||||
if (
|
||||
!value ||
|
||||
value.version !== 1 ||
|
||||
!/^[a-f0-9]{32}$/.test(value.nonce || "") ||
|
||||
!candidate.endsWith("-" + value.nonce) ||
|
||||
!["begun", "applying", "applied", "committed"].includes(value.phase)
|
||||
) {
|
||||
throw new Error("invalid accepted workspace transaction phase");
|
||||
}
|
||||
return value.phase;
|
||||
}
|
||||
function transitionPhase(candidate, current, expected, next) {
|
||||
const allowed =
|
||||
(expected === null && next !== null) ||
|
||||
(expected === "begun" && next === "applying") ||
|
||||
(expected === "applying" && next === "applied") ||
|
||||
(expected === "applied" && next === "committed");
|
||||
if (current !== expected || !allowed) {
|
||||
throw new Error("invalid accepted workspace transaction phase transition");
|
||||
}
|
||||
const candidateNonce = path.basename(candidate).slice(-32);
|
||||
if (!/^[a-f0-9]{32}$/.test(candidateNonce)) {
|
||||
throw new Error("invalid accepted workspace transaction phase path");
|
||||
}
|
||||
const candidatePhase = path.join(candidate, "phase.json");
|
||||
const temporary = candidatePhase + "." + process.pid + "." + crypto.randomBytes(4).toString("hex");
|
||||
fs.writeFileSync(temporary, JSON.stringify({ version: 1, nonce: candidateNonce, phase: next }), {
|
||||
flag: "wx",
|
||||
mode: 0o600,
|
||||
});
|
||||
fs.renameSync(temporary, candidatePhase);
|
||||
}
|
||||
function normalizeRecoveredPhase(candidate, cleanupNamespace = false) {
|
||||
const phase = readPhase(candidate, false);
|
||||
if (phase !== null) return phase;
|
||||
const inferred = cleanupNamespace
|
||||
? "committed"
|
||||
: exists(path.join(candidate, "applied"))
|
||||
? "applied"
|
||||
: exists(path.join(candidate, "state.json")) ||
|
||||
exists(path.join(candidate, "ancestor-modes.json"))
|
||||
? "applying"
|
||||
: "begun";
|
||||
// Transactions from pre-phase beta workers are normalized only while the
|
||||
// locked recovery owner is deciding their existing durable rollback state.
|
||||
transitionPhase(candidate, null, null, inferred);
|
||||
return inferred;
|
||||
}
|
||||
function readState(candidate) {
|
||||
const value = JSON.parse(fs.readFileSync(path.join(candidate, "state.json"), "utf8"));
|
||||
if (!Array.isArray(value) || value.length > 25_000) {
|
||||
throw new Error("invalid accepted workspace transaction state");
|
||||
}
|
||||
const relatives = parsePaths(JSON.stringify(value.map((entry) => entry && entry.relative)));
|
||||
if (
|
||||
relatives.length !== value.length ||
|
||||
value.some(
|
||||
(entry, index) =>
|
||||
!entry ||
|
||||
entry.relative !== relatives[index] ||
|
||||
typeof entry.hadLive !== "boolean" ||
|
||||
(entry.directoryMode !== undefined &&
|
||||
(!Number.isInteger(entry.directoryMode) ||
|
||||
entry.directoryMode < 0 ||
|
||||
entry.directoryMode > 0o7777)),
|
||||
)
|
||||
) {
|
||||
throw new Error("invalid accepted workspace transaction state");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
function readAncestorModes(candidate) {
|
||||
const candidateModes = path.join(candidate, "ancestor-modes.json");
|
||||
if (!exists(candidateModes)) return [];
|
||||
const value = JSON.parse(fs.readFileSync(candidateModes, "utf8"));
|
||||
if (!Array.isArray(value) || value.length > 250_000) {
|
||||
throw new Error("invalid accepted workspace ancestor modes");
|
||||
}
|
||||
const seen = new Set();
|
||||
for (const entry of value) {
|
||||
if (
|
||||
!entry ||
|
||||
(entry.relative !== "" && !isSafeRelativePath(entry.relative)) ||
|
||||
seen.has(entry.relative) ||
|
||||
!Number.isInteger(entry.mode) ||
|
||||
entry.mode < 0 ||
|
||||
entry.mode > 0o7777
|
||||
) {
|
||||
throw new Error("invalid accepted workspace ancestor modes");
|
||||
}
|
||||
seen.add(entry.relative);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
function writeAncestorModes(value) {
|
||||
const temporary = ancestorModesFile + ".tmp";
|
||||
fs.writeFileSync(temporary, JSON.stringify(value), { flag: "wx", mode: 0o600 });
|
||||
fs.renameSync(temporary, ancestorModesFile);
|
||||
}
|
||||
function ancestorPaths(paths) {
|
||||
const ancestors = new Set();
|
||||
for (const relative of paths) {
|
||||
const segments = relative.split("/");
|
||||
for (let index = 1; index < segments.length; index += 1) {
|
||||
ancestors.add(segments.slice(0, index).join("/"));
|
||||
}
|
||||
}
|
||||
if (ancestors.size + 1 > 250_000) {
|
||||
throw new Error("accepted workspace transaction has too many ancestors");
|
||||
}
|
||||
return [...ancestors].sort((left, right) => {
|
||||
const depth = left.split("/").length - right.split("/").length;
|
||||
return depth || (left < right ? -1 : left > right ? 1 : 0);
|
||||
});
|
||||
}
|
||||
function prepareWritableAncestors(paths) {
|
||||
// parsePaths removes descendants of changed directories, so these are all
|
||||
// unchanged live ancestors. Read every mode before mutating any permission.
|
||||
const modes = ["", ...ancestorPaths(paths)].map((relative) => {
|
||||
const target = relative ? targetPath(root, relative) : root;
|
||||
const stats = fs.lstatSync(target);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
return { relative, mode: stats.mode & 0o7777 };
|
||||
});
|
||||
writeAncestorModes(modes);
|
||||
makeAncestorsWritable(modes);
|
||||
return modes;
|
||||
}
|
||||
function makeAncestorsWritable(modes) {
|
||||
const widened = [];
|
||||
try {
|
||||
for (const entry of modes) {
|
||||
const target = entry.relative ? targetPath(root, entry.relative) : root;
|
||||
const stats = fs.lstatSync(target);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
const currentMode = stats.mode & 0o7777;
|
||||
const writableMode = entry.mode | 0o700;
|
||||
if (currentMode !== writableMode) {
|
||||
fs.chmodSync(target, writableMode);
|
||||
widened.push(entry);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
try {
|
||||
restoreAncestorModes(widened);
|
||||
} catch (restoreError) {
|
||||
const failure = new Error("accepted workspace ancestor mode rollback failed", {
|
||||
cause: error,
|
||||
});
|
||||
Object.defineProperty(failure, "restoreFailure", { value: restoreError });
|
||||
throw failure;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
function restoreAncestorModes(modes) {
|
||||
for (const entry of [...modes].reverse()) {
|
||||
const target = entry.relative ? targetPath(root, entry.relative) : root;
|
||||
const stats = fs.lstatSync(target);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
if ((stats.mode & 0o7777) !== entry.mode) fs.chmodSync(target, entry.mode);
|
||||
}
|
||||
}
|
||||
function removeTransaction(candidate = transaction) {
|
||||
removeTree(candidate);
|
||||
}
|
||||
function restoreTransaction(candidate) {
|
||||
if (!exists(candidate)) return;
|
||||
const ancestorModes = readAncestorModes(candidate);
|
||||
makeAncestorsWritable(ancestorModes);
|
||||
const candidateState = path.join(candidate, "state.json");
|
||||
try {
|
||||
if (exists(candidateState)) {
|
||||
const candidateBackup = path.join(candidate, "backup");
|
||||
for (const entry of [...readState(candidate)].reverse()) {
|
||||
const live = livePath(entry.relative);
|
||||
const backup = targetPath(candidateBackup, entry.relative);
|
||||
if (exists(backup)) {
|
||||
removeTree(live);
|
||||
fs.renameSync(backup, live);
|
||||
if (entry.directoryMode !== undefined) fs.chmodSync(live, entry.directoryMode);
|
||||
} else if (!entry.hadLive) {
|
||||
removeTree(live);
|
||||
} else if (entry.directoryMode !== undefined && exists(live)) {
|
||||
fs.chmodSync(live, entry.directoryMode);
|
||||
}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
restoreAncestorModes(ancestorModes);
|
||||
}
|
||||
removeTransaction(candidate);
|
||||
}
|
||||
function recoverTransaction(candidate) {
|
||||
const phase = normalizeRecoveredPhase(candidate);
|
||||
if (phase === "committed") {
|
||||
throw new Error("committed accepted workspace transaction is outside cleanup");
|
||||
}
|
||||
restoreTransaction(candidate);
|
||||
}
|
||||
function recoverCleanup(candidate) {
|
||||
const phase = normalizeRecoveredPhase(candidate, true);
|
||||
if (phase === "applied") transitionPhase(candidate, phase, "applied", "committed");
|
||||
else if (phase !== "committed") throw new Error("invalid accepted workspace cleanup phase");
|
||||
removeTransaction(candidate);
|
||||
}
|
||||
function recoverTransactions() {
|
||||
for (const name of fs.readdirSync(transactionRoot)) {
|
||||
if (name.startsWith(cleanupPrefix) && /^[a-f0-9]{32}$/.test(name.slice(cleanupPrefix.length))) {
|
||||
recoverCleanup(path.join(transactionRoot, name));
|
||||
}
|
||||
}
|
||||
for (const name of fs.readdirSync(transactionRoot)) {
|
||||
if (
|
||||
name.startsWith(transactionPrefix) &&
|
||||
/^[a-f0-9]{32}$/.test(name.slice(transactionPrefix.length))
|
||||
) {
|
||||
recoverTransaction(path.join(transactionRoot, name));
|
||||
}
|
||||
}
|
||||
}
|
||||
function writeSettlementOutcome(outcome) {
|
||||
process.stdout.write(JSON.stringify({ version: 1, outcome }) + "\n");
|
||||
}
|
||||
function runAction() {
|
||||
if (action === "begin") {
|
||||
const paths = parsePaths(fs.readFileSync(0, "utf8"));
|
||||
recoverTransactions();
|
||||
fs.mkdirSync(transaction, { mode: 0o700 });
|
||||
fs.mkdirSync(nextRoot, { mode: 0o700 });
|
||||
fs.mkdirSync(backupRoot, { mode: 0o700 });
|
||||
fs.writeFileSync(pathsFile, JSON.stringify(paths), { mode: 0o600 });
|
||||
transitionPhase(transaction, null, null, "begun");
|
||||
process.stdout.write(nextRoot + "\n");
|
||||
return;
|
||||
}
|
||||
if (action === "apply") {
|
||||
const phase = readPhase(transaction);
|
||||
if (phase === "applied") return;
|
||||
if (phase === "applying") {
|
||||
restoreTransaction(transaction);
|
||||
throw new Error("recovered interrupted accepted workspace apply");
|
||||
}
|
||||
if (phase !== "begun") throw new Error("accepted workspace transaction cannot be applied");
|
||||
transitionPhase(transaction, phase, "begun", "applying");
|
||||
const paths = readPaths();
|
||||
try {
|
||||
const ancestorModes = prepareWritableAncestors(paths);
|
||||
const state = paths.map((relative) => {
|
||||
const live = livePath(relative);
|
||||
if (!exists(live)) return { relative, hadLive: false };
|
||||
const stats = fs.lstatSync(live);
|
||||
return {
|
||||
relative,
|
||||
hadLive: true,
|
||||
...(stats.isDirectory() && !stats.isSymbolicLink()
|
||||
? { directoryMode: stats.mode & 0o7777 }
|
||||
: {}),
|
||||
};
|
||||
});
|
||||
const temporaryStateFile = stateFile + ".tmp";
|
||||
fs.writeFileSync(temporaryStateFile, JSON.stringify(state), { flag: "wx", mode: 0o600 });
|
||||
fs.renameSync(temporaryStateFile, stateFile);
|
||||
for (const entry of state) {
|
||||
if (!entry.hadLive) continue;
|
||||
const source = livePath(entry.relative);
|
||||
const sourceStats = fs.lstatSync(source);
|
||||
const destination = targetPath(backupRoot, entry.relative);
|
||||
fs.mkdirSync(path.dirname(destination), { recursive: true, mode: 0o700 });
|
||||
try {
|
||||
if (sourceStats.isDirectory() && !sourceStats.isSymbolicLink()) {
|
||||
fs.chmodSync(source, 0o700);
|
||||
}
|
||||
fs.renameSync(source, destination);
|
||||
} catch (error) {
|
||||
if (entry.directoryMode !== undefined && exists(source)) {
|
||||
fs.chmodSync(source, entry.directoryMode);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
for (const entry of state) {
|
||||
const source = targetPath(nextRoot, entry.relative);
|
||||
if (exists(source)) fs.renameSync(source, livePath(entry.relative));
|
||||
}
|
||||
restoreAncestorModes(ancestorModes);
|
||||
transitionPhase(transaction, "applying", "applying", "applied");
|
||||
} catch (error) {
|
||||
restoreTransaction(transaction);
|
||||
throw error;
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (action === "rollback") {
|
||||
if (exists(cleanup)) {
|
||||
if (exists(transaction)) throw new Error("ambiguous accepted workspace transaction state");
|
||||
const cleanupPhase = normalizeRecoveredPhase(cleanup, true);
|
||||
if (cleanupPhase !== "applied" && cleanupPhase !== "committed") {
|
||||
throw new Error("accepted workspace cleanup cannot be rolled back");
|
||||
}
|
||||
fs.renameSync(cleanup, transaction);
|
||||
restoreTransaction(transaction);
|
||||
} else if (exists(transaction)) {
|
||||
recoverTransaction(transaction);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (action === "recover") {
|
||||
recoverTransactions();
|
||||
return;
|
||||
}
|
||||
if (action === "settle") {
|
||||
if (exists(transaction) && exists(cleanup)) {
|
||||
throw new Error("ambiguous accepted workspace transaction state");
|
||||
}
|
||||
if (exists(cleanup)) {
|
||||
const phase = normalizeRecoveredPhase(cleanup, true);
|
||||
if (phase === "applied") transitionPhase(cleanup, phase, "applied", "committed");
|
||||
else if (phase !== "committed") throw new Error("invalid accepted workspace cleanup phase");
|
||||
writeSettlementOutcome("committed");
|
||||
return;
|
||||
}
|
||||
if (!exists(transaction)) {
|
||||
writeSettlementOutcome("rolled-back");
|
||||
return;
|
||||
}
|
||||
const phase = normalizeRecoveredPhase(transaction);
|
||||
if (phase === "applied") {
|
||||
writeSettlementOutcome("applied");
|
||||
return;
|
||||
}
|
||||
if (phase === "applying") {
|
||||
restoreTransaction(transaction);
|
||||
writeSettlementOutcome("rolled-back");
|
||||
return;
|
||||
}
|
||||
if (phase === "begun") {
|
||||
writeSettlementOutcome("begun");
|
||||
return;
|
||||
}
|
||||
throw new Error("invalid accepted workspace settlement phase");
|
||||
}
|
||||
if (action === "commit") {
|
||||
if (exists(transaction) && exists(cleanup)) {
|
||||
throw new Error("ambiguous accepted workspace transaction state");
|
||||
}
|
||||
if (exists(cleanup)) {
|
||||
const phase = readPhase(cleanup);
|
||||
if (phase === "applied") transitionPhase(cleanup, phase, "applied", "committed");
|
||||
else if (phase !== "committed") throw new Error("accepted workspace cleanup is not committed");
|
||||
} else if (exists(transaction)) {
|
||||
const phase = readPhase(transaction);
|
||||
if (phase !== "applied") throw new Error("accepted workspace transaction is not applied");
|
||||
// The namespace rename is the commit point. Later recovery removes the backup
|
||||
// only after the gateway has had a chance to observe this command's success.
|
||||
fs.renameSync(transaction, cleanup);
|
||||
transitionPhase(cleanup, phase, "applied", "committed");
|
||||
}
|
||||
return;
|
||||
}
|
||||
throw new Error("invalid accepted workspace transaction action");
|
||||
}
|
||||
// Every mutating action and SSH-loss settlement shares this remote owner lock;
|
||||
// a disconnected gateway can never overlap rollback with the live apply process.
|
||||
acquireWorkspaceLock();
|
||||
try {
|
||||
runAction();
|
||||
} finally {
|
||||
releaseWorkspaceLock();
|
||||
}`;
|
||||
@@ -0,0 +1,701 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { waitForChildClose, waitForFile } from "../../../test/helpers/process-wait.js";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
|
||||
import { runCommandWithTimeout, type SpawnResult } from "../../process/exec.js";
|
||||
import { createDeferred } from "../../shared/deferred.js";
|
||||
import {
|
||||
WorkerTunnelOwnerDisconnectedError,
|
||||
type WorkerWorkspaceCommand,
|
||||
} from "./tunnel-contract.js";
|
||||
import {
|
||||
AcceptedWorkspacePublicationIndeterminateError,
|
||||
isAcceptedWorkspacePublicationIndeterminateError,
|
||||
} from "./workspace-accepted-publication.js";
|
||||
import {
|
||||
createAcceptedWorkspacePublisherFactory,
|
||||
recoverAcceptedWorkspacePublication,
|
||||
} from "./workspace-accepted-sync.js";
|
||||
import {
|
||||
serializeWorkerWorkspaceManifest,
|
||||
type WorkerWorkspaceManifest,
|
||||
} from "./workspace-manifest.js";
|
||||
import {
|
||||
REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS,
|
||||
REMOTE_WORKSPACE_MANIFEST_JS,
|
||||
} from "./workspace-sync-scripts.js";
|
||||
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
function result(overrides: Partial<SpawnResult> = {}): SpawnResult {
|
||||
return {
|
||||
stdout: "",
|
||||
stderr: "",
|
||||
code: 0,
|
||||
signal: null,
|
||||
killed: false,
|
||||
termination: "exit",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function manifest(content: string): WorkerWorkspaceManifest {
|
||||
return {
|
||||
version: 1,
|
||||
baseCommit: null,
|
||||
entries: [
|
||||
{
|
||||
path: "result.txt",
|
||||
type: "file",
|
||||
mode: 0o644,
|
||||
size: Buffer.byteLength(content),
|
||||
sha256: createHash("sha256").update(content).digest("hex"),
|
||||
},
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
function manifestRef(value: WorkerWorkspaceManifest): string {
|
||||
return `sha256:${createHash("sha256").update(serializeWorkerWorkspaceManifest(value)).digest("hex")}`;
|
||||
}
|
||||
|
||||
function settlement(outcome: "begun" | "rolled-back" | "applied" | "committed"): SpawnResult {
|
||||
return result({ stdout: `${JSON.stringify({ version: 1, outcome })}\n` });
|
||||
}
|
||||
|
||||
describe("accepted workspace publication", () => {
|
||||
it("settles a still-running apply after SSH loses its exit status", async () => {
|
||||
const root = tempDirs.make("openclaw-accepted-ssh-loss-");
|
||||
const local = path.join(root, "local");
|
||||
let workspace = path.join(root, "workspace");
|
||||
const gate = path.join(root, "gate.fifo");
|
||||
const applyMarker = path.join(root, "apply-started");
|
||||
const settleStarted = createDeferred();
|
||||
const preload = path.join(root, "gate.cjs");
|
||||
await Promise.all([fs.mkdir(local), fs.mkdir(workspace)]);
|
||||
workspace = await fs.realpath(workspace);
|
||||
await Promise.all([
|
||||
fs.writeFile(path.join(local, "result.txt"), "local\n"),
|
||||
fs.writeFile(path.join(workspace, "result.txt"), "worker\n"),
|
||||
]);
|
||||
expect((await runCommandWithTimeout(["mkfifo", gate], { timeoutMs: 10_000 })).code).toBe(0);
|
||||
await fs.writeFile(
|
||||
preload,
|
||||
`const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const renameSync = fs.renameSync;
|
||||
let gated = false;
|
||||
fs.renameSync = function(source, destination) {
|
||||
const value = renameSync.apply(this, arguments);
|
||||
if (!gated && process.argv[1] === "apply" && source === process.env.OPENCLAW_TEST_GATE_SOURCE && destination.includes(path.sep + "backup" + path.sep)) {
|
||||
gated = true;
|
||||
fs.writeFileSync(process.env.OPENCLAW_TEST_APPLY_MARKER, "");
|
||||
fs.readFileSync(process.env.OPENCLAW_TEST_GATE);
|
||||
}
|
||||
return value;
|
||||
};
|
||||
`,
|
||||
);
|
||||
const env = {
|
||||
...process.env,
|
||||
OPENCLAW_TEST_GATE: gate,
|
||||
OPENCLAW_TEST_GATE_SOURCE: path.join(workspace, "result.txt"),
|
||||
OPENCLAW_TEST_APPLY_MARKER: applyMarker,
|
||||
};
|
||||
const remote = manifest("worker\n");
|
||||
const accepted = manifest("local\n");
|
||||
const acceptedRef = manifestRef(accepted);
|
||||
const transactionCalls: Array<{
|
||||
action: string;
|
||||
nonce: string;
|
||||
transportRetry: WorkerWorkspaceCommand["transportRetry"];
|
||||
}> = [];
|
||||
const manifestCalls: Array<WorkerWorkspaceCommand["transportRetry"]> = [];
|
||||
let stagingRoot: string | undefined;
|
||||
let applyExited:
|
||||
| Promise<{ code: number | null; signal: NodeJS.Signals | null; stderr: string }>
|
||||
| undefined;
|
||||
const runWorkspaceCommand = async (command: WorkerWorkspaceCommand): Promise<SpawnResult> => {
|
||||
const transactionAction =
|
||||
command.argv[2] === REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS ? command.argv[3] : undefined;
|
||||
if (!transactionAction) {
|
||||
expect(command.argv[2]).toBe(REMOTE_WORKSPACE_MANIFEST_JS);
|
||||
manifestCalls.push(command.transportRetry);
|
||||
return result({ stdout: command.argv[5] === "publish" ? "" : `${acceptedRef}\n` });
|
||||
}
|
||||
transactionCalls.push({
|
||||
action: transactionAction,
|
||||
nonce: command.argv[5]!,
|
||||
transportRetry: command.transportRetry,
|
||||
});
|
||||
if (transactionAction === "settle") {
|
||||
settleStarted.resolve();
|
||||
}
|
||||
if (transactionAction === "apply") {
|
||||
const child = spawn(process.execPath, ["--require", preload, ...command.argv.slice(1)], {
|
||||
env,
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
});
|
||||
child.stdin.end(command.input);
|
||||
let stderr = "";
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stderr.on("data", (chunk: string) => {
|
||||
stderr += chunk;
|
||||
});
|
||||
applyExited = waitForChildClose(child, 10_000).then(({ code, signal }) => ({
|
||||
code,
|
||||
signal,
|
||||
stderr,
|
||||
}));
|
||||
await waitForFile(applyMarker, 10_000);
|
||||
return result({ code: 255, stderr: "connection lost after remote apply started" });
|
||||
}
|
||||
const commandResult = await runCommandWithTimeout(
|
||||
[process.execPath, ...command.argv.slice(1)],
|
||||
{
|
||||
timeoutMs: 10_000,
|
||||
baseEnv: env,
|
||||
input: command.input,
|
||||
},
|
||||
);
|
||||
if (transactionAction === "begin" && commandResult.code === 0) {
|
||||
stagingRoot = commandResult.stdout.trim();
|
||||
}
|
||||
return commandResult;
|
||||
};
|
||||
const runRsync = async (): Promise<SpawnResult> => {
|
||||
if (!stagingRoot) {
|
||||
throw new Error("accepted transaction did not begin before transfer");
|
||||
}
|
||||
await fs.copyFile(path.join(local, "result.txt"), path.join(stagingRoot, "result.txt"));
|
||||
return result();
|
||||
};
|
||||
const publisher = createAcceptedWorkspacePublisherFactory({
|
||||
runWorkspaceCommand,
|
||||
runRsync,
|
||||
scpTarget: "test",
|
||||
localPath: local,
|
||||
remoteWorkspaceDir: workspace,
|
||||
})(remote, manifestRef(remote));
|
||||
|
||||
const publishing = publisher.publishAcceptedManifest({
|
||||
manifestRef: acceptedRef,
|
||||
manifest: accepted,
|
||||
conflictPaths: ["result.txt"],
|
||||
});
|
||||
let publishingSettled = false;
|
||||
void publishing.then(
|
||||
() => {
|
||||
publishingSettled = true;
|
||||
},
|
||||
() => {
|
||||
publishingSettled = true;
|
||||
},
|
||||
);
|
||||
await waitForFile(applyMarker, 10_000);
|
||||
await settleStarted.promise;
|
||||
expect(transactionCalls.map((entry) => entry.action)).toEqual(["begin", "apply", "settle"]);
|
||||
expect(new Set(transactionCalls.map((entry) => entry.nonce)).size).toBe(1);
|
||||
expect(transactionCalls.every((entry) => entry.transportRetry === "never")).toBe(true);
|
||||
expect(manifestCalls).toEqual(["idempotent"]);
|
||||
expect(publishingSettled).toBe(false);
|
||||
await expect(fs.access(path.join(workspace, "result.txt"))).rejects.toThrow();
|
||||
expect(transactionCalls.some((entry) => entry.action === "rollback")).toBe(false);
|
||||
|
||||
const gateWriter = await fs.open(gate, "w");
|
||||
await gateWriter.write("release");
|
||||
await gateWriter.close();
|
||||
await expect(publishing).resolves.toBeUndefined();
|
||||
if (!applyExited) {
|
||||
throw new Error("remote apply process was not started");
|
||||
}
|
||||
await expect(applyExited).resolves.toMatchObject({ code: 0, signal: null, stderr: "" });
|
||||
expect(transactionCalls.map((entry) => entry.action)).toEqual([
|
||||
"begin",
|
||||
"apply",
|
||||
"settle",
|
||||
"commit",
|
||||
]);
|
||||
expect(new Set(transactionCalls.map((entry) => entry.nonce)).size).toBe(1);
|
||||
expect(transactionCalls.every((entry) => entry.transportRetry === "never")).toBe(true);
|
||||
expect(manifestCalls).toEqual(["idempotent", "idempotent"]);
|
||||
await expect(fs.readFile(path.join(workspace, "result.txt"), "utf8")).resolves.toBe("local\n");
|
||||
await expect(fs.readFile(path.join(local, "result.txt"), "utf8")).resolves.toBe("local\n");
|
||||
|
||||
await recoverAcceptedWorkspacePublication({
|
||||
runWorkspaceCommand,
|
||||
remoteWorkspaceDir: workspace,
|
||||
});
|
||||
expect(transactionCalls.map((entry) => entry.action)).toEqual([
|
||||
"begin",
|
||||
"apply",
|
||||
"settle",
|
||||
"commit",
|
||||
"recover",
|
||||
]);
|
||||
expect(transactionCalls.every((entry) => entry.transportRetry === "never")).toBe(true);
|
||||
expect(
|
||||
(await fs.readdir(root)).filter((name) => name.startsWith(".openclaw-accepted-")),
|
||||
).toEqual([]);
|
||||
});
|
||||
|
||||
it("leaves publication pending when settle reaches its lock deadline behind a live apply", async () => {
|
||||
const root = tempDirs.make("openclaw-accepted-settle-deadline-");
|
||||
const local = path.join(root, "local");
|
||||
let workspace = path.join(root, "workspace");
|
||||
const gate = path.join(root, "gate.fifo");
|
||||
const applyMarker = path.join(root, "apply-started");
|
||||
const applyPreload = path.join(root, "apply-gate.cjs");
|
||||
const settlePreload = path.join(root, "settle-clock.cjs");
|
||||
await Promise.all([fs.mkdir(local), fs.mkdir(workspace)]);
|
||||
workspace = await fs.realpath(workspace);
|
||||
await Promise.all([
|
||||
fs.writeFile(path.join(local, "result.txt"), "local\n"),
|
||||
fs.writeFile(path.join(workspace, "result.txt"), "worker\n"),
|
||||
]);
|
||||
expect((await runCommandWithTimeout(["mkfifo", gate], { timeoutMs: 10_000 })).code).toBe(0);
|
||||
const gateController = await fs.open(gate, "r+");
|
||||
let gateReleased = false;
|
||||
const releaseApply = async () => {
|
||||
if (gateReleased) {
|
||||
return;
|
||||
}
|
||||
gateReleased = true;
|
||||
await gateController.write("release");
|
||||
await gateController.close();
|
||||
};
|
||||
await Promise.all([
|
||||
fs.writeFile(
|
||||
applyPreload,
|
||||
`const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const renameSync = fs.renameSync;
|
||||
let gated = false;
|
||||
fs.renameSync = function(source, destination) {
|
||||
const value = renameSync.apply(this, arguments);
|
||||
if (!gated && process.argv[1] === "apply" && source === process.env.OPENCLAW_TEST_GATE_SOURCE && destination.includes(path.sep + "backup" + path.sep)) {
|
||||
gated = true;
|
||||
fs.writeFileSync(process.env.OPENCLAW_TEST_APPLY_MARKER, "");
|
||||
fs.readFileSync(process.env.OPENCLAW_TEST_GATE);
|
||||
}
|
||||
return value;
|
||||
};
|
||||
`,
|
||||
),
|
||||
fs.writeFile(
|
||||
settlePreload,
|
||||
`let now = 0;
|
||||
Date.now = () => now;
|
||||
Atomics.wait = function(waitArray, index, value, timeout) {
|
||||
now += 9 * 60 * 1000 + Number(timeout || 0) + 1;
|
||||
return "timed-out";
|
||||
};
|
||||
`,
|
||||
),
|
||||
]);
|
||||
const env = {
|
||||
...process.env,
|
||||
OPENCLAW_TEST_GATE: gate,
|
||||
OPENCLAW_TEST_GATE_SOURCE: path.join(workspace, "result.txt"),
|
||||
OPENCLAW_TEST_APPLY_MARKER: applyMarker,
|
||||
};
|
||||
const remote = manifest("worker\n");
|
||||
const accepted = manifest("local\n");
|
||||
const transactionCalls: Array<{ action: string; nonce: string }> = [];
|
||||
let stagingRoot: string | undefined;
|
||||
let applyChild: ReturnType<typeof spawn> | undefined;
|
||||
let applyExited:
|
||||
| Promise<{ code: number | null; signal: NodeJS.Signals | null; stderr: string }>
|
||||
| undefined;
|
||||
const runWorkspaceCommand = async (command: WorkerWorkspaceCommand): Promise<SpawnResult> => {
|
||||
const action =
|
||||
command.argv[2] === REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS ? command.argv[3] : undefined;
|
||||
if (!action) {
|
||||
return result();
|
||||
}
|
||||
transactionCalls.push({ action, nonce: command.argv[5]! });
|
||||
if (action === "apply") {
|
||||
const child = spawn(
|
||||
process.execPath,
|
||||
["--require", applyPreload, ...command.argv.slice(1)],
|
||||
{ env, stdio: ["pipe", "pipe", "pipe"] },
|
||||
);
|
||||
applyChild = child;
|
||||
child.stdin.end(command.input);
|
||||
let stderr = "";
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stderr.on("data", (chunk: string) => {
|
||||
stderr += chunk;
|
||||
});
|
||||
applyExited = waitForChildClose(child, 10_000).then(({ code, signal }) => ({
|
||||
code,
|
||||
signal,
|
||||
stderr,
|
||||
}));
|
||||
await waitForFile(applyMarker, 10_000);
|
||||
return result({ code: 255, stderr: "connection lost after remote apply started" });
|
||||
}
|
||||
const commandResult = await runCommandWithTimeout(
|
||||
[
|
||||
process.execPath,
|
||||
...(action === "settle" ? ["--require", settlePreload] : []),
|
||||
...command.argv.slice(1),
|
||||
],
|
||||
{ timeoutMs: 10_000, baseEnv: env, input: command.input },
|
||||
);
|
||||
if (action === "begin" && commandResult.code === 0) {
|
||||
stagingRoot = commandResult.stdout.trim();
|
||||
}
|
||||
return commandResult;
|
||||
};
|
||||
const publisher = createAcceptedWorkspacePublisherFactory({
|
||||
runWorkspaceCommand,
|
||||
runRsync: async () => {
|
||||
if (!stagingRoot) {
|
||||
throw new Error("accepted transaction did not begin before transfer");
|
||||
}
|
||||
await fs.copyFile(path.join(local, "result.txt"), path.join(stagingRoot, "result.txt"));
|
||||
return result();
|
||||
},
|
||||
scpTarget: "test",
|
||||
localPath: local,
|
||||
remoteWorkspaceDir: workspace,
|
||||
})(remote, manifestRef(remote));
|
||||
|
||||
try {
|
||||
const thrown = await publisher
|
||||
.publishAcceptedManifest({
|
||||
manifestRef: manifestRef(accepted),
|
||||
manifest: accepted,
|
||||
conflictPaths: ["result.txt"],
|
||||
})
|
||||
.catch((error: unknown) => error);
|
||||
|
||||
expect(thrown).toBeInstanceOf(AcceptedWorkspacePublicationIndeterminateError);
|
||||
expect(transactionCalls.map(({ action }) => action)).toEqual(["begin", "apply", "settle"]);
|
||||
expect(new Set(transactionCalls.map(({ nonce }) => nonce)).size).toBe(1);
|
||||
expect(transactionCalls.some(({ action }) => action === "rollback")).toBe(false);
|
||||
await expect(fs.access(path.join(workspace, "result.txt"))).rejects.toThrow();
|
||||
|
||||
await releaseApply();
|
||||
if (!applyExited) {
|
||||
throw new Error("remote apply process was not started");
|
||||
}
|
||||
await expect(applyExited).resolves.toMatchObject({ code: 0, signal: null, stderr: "" });
|
||||
await expect(fs.readFile(path.join(workspace, "result.txt"), "utf8")).resolves.toBe(
|
||||
"local\n",
|
||||
);
|
||||
|
||||
await recoverAcceptedWorkspacePublication({
|
||||
runWorkspaceCommand,
|
||||
remoteWorkspaceDir: workspace,
|
||||
});
|
||||
await expect(fs.readFile(path.join(workspace, "result.txt"), "utf8")).resolves.toBe(
|
||||
"worker\n",
|
||||
);
|
||||
expect(
|
||||
(await fs.readdir(root)).filter((name) => name.startsWith(".openclaw-accepted-")),
|
||||
).toEqual([]);
|
||||
} finally {
|
||||
await releaseApply().catch(() => undefined);
|
||||
await applyExited?.catch(async () => {
|
||||
if (applyChild?.exitCode === null && applyChild.signalCode === null) {
|
||||
applyChild.kill("SIGTERM");
|
||||
await waitForChildClose(applyChild, 1_000).catch(() => undefined);
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps an unobservable apply pending when settlement is unobservable", async () => {
|
||||
const remote = manifest("worker\n");
|
||||
const accepted = manifest("local\n");
|
||||
const actions: string[] = [];
|
||||
const factory = createAcceptedWorkspacePublisherFactory({
|
||||
runWorkspaceCommand: async (command) => {
|
||||
if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) {
|
||||
return result();
|
||||
}
|
||||
const action = command.argv[3];
|
||||
actions.push(action!);
|
||||
if (action === "begin") {
|
||||
return result({ stdout: "/remote/staging\n" });
|
||||
}
|
||||
if (action === "apply") {
|
||||
return result({ code: 255, stderr: "apply transport lost" });
|
||||
}
|
||||
if (action === "settle") {
|
||||
throw new WorkerTunnelOwnerDisconnectedError();
|
||||
}
|
||||
return result();
|
||||
},
|
||||
runRsync: async () => result(),
|
||||
scpTarget: "test",
|
||||
localPath: "/local",
|
||||
remoteWorkspaceDir: "/remote",
|
||||
});
|
||||
|
||||
const publishing = factory(remote, manifestRef(remote)).publishAcceptedManifest({
|
||||
manifestRef: manifestRef(accepted),
|
||||
manifest: accepted,
|
||||
conflictPaths: ["result.txt"],
|
||||
});
|
||||
const thrown = await publishing.catch((error: unknown) => error);
|
||||
expect(thrown).toBeInstanceOf(AcceptedWorkspacePublicationIndeterminateError);
|
||||
expect(isAcceptedWorkspacePublicationIndeterminateError(thrown)).toBe(true);
|
||||
expect(thrown).toMatchObject({
|
||||
message: "Accepted workspace publication is indeterminate and requires recovery",
|
||||
operation: "apply",
|
||||
cause: expect.objectContaining({
|
||||
message: "Worker workspace sync failed: apply transport lost",
|
||||
}),
|
||||
observationFailure: expect.any(WorkerTunnelOwnerDisconnectedError),
|
||||
});
|
||||
expect(actions).toEqual(["begin", "apply", "settle"]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
name: "an ordinary observed failure",
|
||||
apply: result({ code: 1, stderr: "apply rejected" }),
|
||||
settle: undefined,
|
||||
actions: ["begin", "apply", "rollback"],
|
||||
},
|
||||
{
|
||||
name: "settlement observes the begun phase",
|
||||
apply: result({ code: 255, stderr: "apply transport lost" }),
|
||||
settle: settlement("begun"),
|
||||
actions: ["begin", "apply", "settle", "rollback"],
|
||||
},
|
||||
{
|
||||
name: "settlement observes a completed rollback",
|
||||
apply: result({ code: 255, stderr: "apply transport lost" }),
|
||||
settle: settlement("rolled-back"),
|
||||
actions: ["begin", "apply", "settle", "rollback"],
|
||||
},
|
||||
])("uses the safe rollback path after $name", async ({ apply, settle, actions: expected }) => {
|
||||
const remote = manifest("worker\n");
|
||||
const accepted = manifest("local\n");
|
||||
const actions: string[] = [];
|
||||
const publisher = createAcceptedWorkspacePublisherFactory({
|
||||
runWorkspaceCommand: async (command) => {
|
||||
if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) {
|
||||
return result();
|
||||
}
|
||||
const action = command.argv[3]!;
|
||||
actions.push(action);
|
||||
if (action === "begin") {
|
||||
return result({ stdout: "/remote/staging\n" });
|
||||
}
|
||||
if (action === "apply") {
|
||||
return apply;
|
||||
}
|
||||
if (action === "settle") {
|
||||
if (!settle) {
|
||||
throw new Error("unexpected settlement");
|
||||
}
|
||||
return settle;
|
||||
}
|
||||
return result();
|
||||
},
|
||||
runRsync: async () => result(),
|
||||
scpTarget: "test",
|
||||
localPath: "/local",
|
||||
remoteWorkspaceDir: "/remote",
|
||||
})(remote, manifestRef(remote));
|
||||
|
||||
await expect(
|
||||
publisher.publishAcceptedManifest({
|
||||
manifestRef: manifestRef(accepted),
|
||||
manifest: accepted,
|
||||
conflictPaths: ["result.txt"],
|
||||
}),
|
||||
).rejects.toThrow(/apply (?:rejected|transport lost)/u);
|
||||
expect(actions).toEqual(expected);
|
||||
});
|
||||
|
||||
it("keeps an ambiguous apply pending when settlement output is malformed", async () => {
|
||||
const remote = manifest("worker\n");
|
||||
const accepted = manifest("local\n");
|
||||
const actions: string[] = [];
|
||||
const publisher = createAcceptedWorkspacePublisherFactory({
|
||||
runWorkspaceCommand: async (command) => {
|
||||
if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) {
|
||||
return result();
|
||||
}
|
||||
const action = command.argv[3]!;
|
||||
actions.push(action);
|
||||
if (action === "begin") {
|
||||
return result({ stdout: "/remote/staging\n" });
|
||||
}
|
||||
if (action === "apply") {
|
||||
return result({ code: 255, stderr: "apply transport lost" });
|
||||
}
|
||||
if (action === "settle") {
|
||||
return result({ stdout: '{"version":1,"outcome":"applied","extra":true}\n' });
|
||||
}
|
||||
return result();
|
||||
},
|
||||
runRsync: async () => result(),
|
||||
scpTarget: "test",
|
||||
localPath: "/local",
|
||||
remoteWorkspaceDir: "/remote",
|
||||
})(remote, manifestRef(remote));
|
||||
|
||||
await expect(
|
||||
publisher.publishAcceptedManifest({
|
||||
manifestRef: manifestRef(accepted),
|
||||
manifest: accepted,
|
||||
conflictPaths: ["result.txt"],
|
||||
}),
|
||||
).rejects.toBeInstanceOf(AcceptedWorkspacePublicationIndeterminateError);
|
||||
expect(actions).toEqual(["begin", "apply", "settle"]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
name: "settlement observes the committed phase",
|
||||
outcome: "committed" as const,
|
||||
retry: false,
|
||||
secondCommit: result(),
|
||||
expected: "success" as const,
|
||||
},
|
||||
{
|
||||
name: "settlement observes applied and retries commit once",
|
||||
outcome: "applied" as const,
|
||||
retry: true,
|
||||
secondCommit: result(),
|
||||
expected: "success" as const,
|
||||
},
|
||||
{
|
||||
name: "the one safe commit retry is unobservable",
|
||||
outcome: "applied" as const,
|
||||
retry: true,
|
||||
secondCommit: result({ code: 255, stderr: "retry transport lost" }),
|
||||
expected: "pending" as const,
|
||||
},
|
||||
{
|
||||
name: "settlement observes the begun phase",
|
||||
outcome: "begun" as const,
|
||||
retry: false,
|
||||
secondCommit: result(),
|
||||
expected: "rollback" as const,
|
||||
},
|
||||
{
|
||||
name: "settlement observes a completed rollback",
|
||||
outcome: "rolled-back" as const,
|
||||
retry: false,
|
||||
secondCommit: result(),
|
||||
expected: "rollback" as const,
|
||||
},
|
||||
{
|
||||
name: "the one safe commit retry is rejected",
|
||||
outcome: "applied" as const,
|
||||
retry: true,
|
||||
secondCommit: result({ code: 1, stderr: "retry rejected" }),
|
||||
expected: "rollback" as const,
|
||||
},
|
||||
])(
|
||||
"handles an ambiguous commit when $name",
|
||||
async ({ outcome, retry, secondCommit, expected }) => {
|
||||
const remote = manifest("worker\n");
|
||||
const accepted = manifest("local\n");
|
||||
const acceptedRef = manifestRef(accepted);
|
||||
const transactionCalls: Array<{ action: string; nonce: string }> = [];
|
||||
let commitCount = 0;
|
||||
const publisher = createAcceptedWorkspacePublisherFactory({
|
||||
runWorkspaceCommand: async (command) => {
|
||||
if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) {
|
||||
return result({ stdout: command.argv[5] === "publish" ? "" : `${acceptedRef}\n` });
|
||||
}
|
||||
const action = command.argv[3]!;
|
||||
transactionCalls.push({ action, nonce: command.argv[5]! });
|
||||
if (action === "begin") {
|
||||
return result({ stdout: "/remote/staging\n" });
|
||||
}
|
||||
if (action === "commit") {
|
||||
commitCount += 1;
|
||||
return commitCount === 1
|
||||
? result({ code: 255, stderr: "commit transport lost" })
|
||||
: secondCommit;
|
||||
}
|
||||
if (action === "settle") {
|
||||
return settlement(outcome);
|
||||
}
|
||||
return result();
|
||||
},
|
||||
runRsync: async () => result(),
|
||||
scpTarget: "test",
|
||||
localPath: "/local",
|
||||
remoteWorkspaceDir: "/remote",
|
||||
})(remote, manifestRef(remote));
|
||||
|
||||
const publishing = publisher.publishAcceptedManifest({
|
||||
manifestRef: acceptedRef,
|
||||
manifest: accepted,
|
||||
conflictPaths: ["result.txt"],
|
||||
});
|
||||
|
||||
if (expected === "success") {
|
||||
await expect(publishing).resolves.toBeUndefined();
|
||||
} else if (expected === "pending") {
|
||||
await expect(publishing).rejects.toBeInstanceOf(
|
||||
AcceptedWorkspacePublicationIndeterminateError,
|
||||
);
|
||||
} else {
|
||||
await expect(publishing).rejects.toThrow(/commit transport lost|retry rejected/u);
|
||||
}
|
||||
expect(transactionCalls.map(({ action }) => action)).toEqual([
|
||||
"begin",
|
||||
"apply",
|
||||
"commit",
|
||||
"settle",
|
||||
...(retry ? ["commit"] : []),
|
||||
...(expected === "rollback" ? ["rollback"] : []),
|
||||
]);
|
||||
expect(new Set(transactionCalls.map(({ nonce }) => nonce)).size).toBe(1);
|
||||
expect(transactionCalls.some(({ action }) => action === "rollback")).toBe(
|
||||
expected === "rollback",
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("rolls back after an ordinary observed commit failure", async () => {
|
||||
const remote = manifest("worker\n");
|
||||
const accepted = manifest("local\n");
|
||||
const acceptedRef = manifestRef(accepted);
|
||||
const actions: string[] = [];
|
||||
const publisher = createAcceptedWorkspacePublisherFactory({
|
||||
runWorkspaceCommand: async (command) => {
|
||||
if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) {
|
||||
return result({ stdout: command.argv[5] === "publish" ? "" : `${acceptedRef}\n` });
|
||||
}
|
||||
const action = command.argv[3]!;
|
||||
actions.push(action);
|
||||
if (action === "begin") {
|
||||
return result({ stdout: "/remote/staging\n" });
|
||||
}
|
||||
if (action === "commit") {
|
||||
return result({ code: 1, stderr: "commit rejected" });
|
||||
}
|
||||
return result();
|
||||
},
|
||||
runRsync: async () => result(),
|
||||
scpTarget: "test",
|
||||
localPath: "/local",
|
||||
remoteWorkspaceDir: "/remote",
|
||||
})(remote, manifestRef(remote));
|
||||
|
||||
await expect(
|
||||
publisher.publishAcceptedManifest({
|
||||
manifestRef: acceptedRef,
|
||||
manifest: accepted,
|
||||
conflictPaths: ["result.txt"],
|
||||
}),
|
||||
).rejects.toThrow("commit rejected");
|
||||
expect(actions).toEqual(["begin", "apply", "commit", "rollback"]);
|
||||
});
|
||||
});
|
||||
@@ -4,6 +4,12 @@ import os from "node:os";
|
||||
import path from "node:path";
|
||||
import type { SpawnResult } from "../../process/exec.js";
|
||||
import type { WorkerWorkspaceCommand } from "./tunnel-contract.js";
|
||||
import {
|
||||
AcceptedWorkspacePublicationIndeterminateError,
|
||||
isAcceptedWorkspacePublicationIndeterminateError,
|
||||
parseAcceptedWorkspaceSettlement,
|
||||
type AcceptedWorkspaceSettlementOutcome,
|
||||
} from "./workspace-accepted-publication.js";
|
||||
import {
|
||||
serializeWorkerWorkspaceManifest,
|
||||
type WorkerWorkspaceManifest,
|
||||
@@ -19,6 +25,18 @@ import {
|
||||
REMOTE_WORKSPACE_MANIFEST_JS,
|
||||
} from "./workspace-sync-scripts.js";
|
||||
|
||||
function isIndeterminateWorkspaceCommandResult(result: SpawnResult): boolean {
|
||||
return result.termination !== "exit" || result.code === 255;
|
||||
}
|
||||
|
||||
function acceptedWorkspaceRollbackError(error: unknown, rollbackFailure: unknown): Error {
|
||||
const rollbackError = new Error("Accepted workspace publication rollback failed", {
|
||||
cause: error,
|
||||
});
|
||||
Object.defineProperty(rollbackError, "rollbackFailure", { value: rollbackFailure });
|
||||
return rollbackError;
|
||||
}
|
||||
|
||||
export async function recoverAcceptedWorkspacePublication(params: {
|
||||
runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise<SpawnResult>;
|
||||
remoteWorkspaceDir: string;
|
||||
@@ -107,7 +125,7 @@ function createAcceptedWorkspacePublisher(params: {
|
||||
}
|
||||
|
||||
const transactionNonce = randomBytes(16).toString("hex");
|
||||
const transactionCommand = async (action: "apply" | "rollback" | "commit") =>
|
||||
const transactionCommand = async (action: "apply" | "rollback" | "commit" | "settle") =>
|
||||
await params.runWorkspaceCommand({
|
||||
transportRetry: "never",
|
||||
argv: [
|
||||
@@ -119,6 +137,67 @@ function createAcceptedWorkspacePublisher(params: {
|
||||
transactionNonce,
|
||||
],
|
||||
});
|
||||
const settleIndeterminatePublication = async (
|
||||
operation: "apply" | "commit",
|
||||
publicationFailure: unknown,
|
||||
): Promise<AcceptedWorkspaceSettlementOutcome> => {
|
||||
let settled: SpawnResult;
|
||||
try {
|
||||
settled = await transactionCommand("settle");
|
||||
} catch (observationFailure) {
|
||||
throw new AcceptedWorkspacePublicationIndeterminateError(
|
||||
operation,
|
||||
publicationFailure,
|
||||
observationFailure,
|
||||
);
|
||||
}
|
||||
if (!workerWorkspaceCommandSucceeded(settled)) {
|
||||
throw new AcceptedWorkspacePublicationIndeterminateError(
|
||||
operation,
|
||||
publicationFailure,
|
||||
workspaceSyncError(settled),
|
||||
);
|
||||
}
|
||||
try {
|
||||
return parseAcceptedWorkspaceSettlement(settled.stdout);
|
||||
} catch (observationFailure) {
|
||||
throw new AcceptedWorkspacePublicationIndeterminateError(
|
||||
operation,
|
||||
publicationFailure,
|
||||
observationFailure,
|
||||
);
|
||||
}
|
||||
};
|
||||
const finishIndeterminateCommit = async (commitFailure: unknown): Promise<void> => {
|
||||
const outcome = await settleIndeterminatePublication("commit", commitFailure);
|
||||
if (outcome === "committed") {
|
||||
return;
|
||||
}
|
||||
if (outcome !== "applied") {
|
||||
throw commitFailure;
|
||||
}
|
||||
let retried: SpawnResult;
|
||||
try {
|
||||
retried = await transactionCommand("commit");
|
||||
} catch (observationFailure) {
|
||||
throw new AcceptedWorkspacePublicationIndeterminateError(
|
||||
"commit",
|
||||
commitFailure,
|
||||
observationFailure,
|
||||
);
|
||||
}
|
||||
if (!workerWorkspaceCommandSucceeded(retried)) {
|
||||
const retryFailure = workspaceSyncError(retried);
|
||||
if (!isIndeterminateWorkspaceCommandResult(retried)) {
|
||||
throw retryFailure;
|
||||
}
|
||||
throw new AcceptedWorkspacePublicationIndeterminateError(
|
||||
"commit",
|
||||
commitFailure,
|
||||
retryFailure,
|
||||
);
|
||||
}
|
||||
};
|
||||
let transactionBegun = false;
|
||||
try {
|
||||
const begun = await params.runWorkspaceCommand({
|
||||
@@ -179,26 +258,55 @@ function createAcceptedWorkspacePublisher(params: {
|
||||
}
|
||||
}
|
||||
|
||||
const applied = await transactionCommand("apply");
|
||||
if (!workerWorkspaceCommandSucceeded(applied)) {
|
||||
throw workspaceSyncError(applied);
|
||||
let applied: SpawnResult | undefined;
|
||||
try {
|
||||
applied = await transactionCommand("apply");
|
||||
} catch (applyFailure) {
|
||||
const outcome = await settleIndeterminatePublication("apply", applyFailure);
|
||||
if (outcome !== "applied" && outcome !== "committed") {
|
||||
throw applyFailure;
|
||||
}
|
||||
}
|
||||
if (applied && !workerWorkspaceCommandSucceeded(applied)) {
|
||||
const applyFailure = workspaceSyncError(applied);
|
||||
if (!isIndeterminateWorkspaceCommandResult(applied)) {
|
||||
throw applyFailure;
|
||||
}
|
||||
const outcome = await settleIndeterminatePublication("apply", applyFailure);
|
||||
if (outcome !== "applied" && outcome !== "committed") {
|
||||
throw applyFailure;
|
||||
}
|
||||
}
|
||||
await verifyAcceptedWorkspace();
|
||||
const committed = await transactionCommand("commit");
|
||||
let committed: SpawnResult;
|
||||
try {
|
||||
committed = await transactionCommand("commit");
|
||||
} catch (commitFailure) {
|
||||
await finishIndeterminateCommit(commitFailure);
|
||||
return;
|
||||
}
|
||||
if (!workerWorkspaceCommandSucceeded(committed)) {
|
||||
throw workspaceSyncError(committed);
|
||||
const commitFailure = workspaceSyncError(committed);
|
||||
if (isIndeterminateWorkspaceCommandResult(committed)) {
|
||||
await finishIndeterminateCommit(commitFailure);
|
||||
return;
|
||||
}
|
||||
throw commitFailure;
|
||||
}
|
||||
} catch (error) {
|
||||
// Transport or settlement timeouts are observation evidence, never authority
|
||||
// for an inverse operation; recovery owns restoring both sides.
|
||||
if (isAcceptedWorkspacePublicationIndeterminateError(error)) {
|
||||
throw error;
|
||||
}
|
||||
if (transactionBegun) {
|
||||
const rolledBack = await transactionCommand("rollback");
|
||||
if (!workerWorkspaceCommandSucceeded(rolledBack)) {
|
||||
const rollbackError = new Error("Accepted workspace publication rollback failed", {
|
||||
cause: error,
|
||||
});
|
||||
Object.defineProperty(rollbackError, "rollbackFailure", {
|
||||
value: workspaceSyncError(rolledBack),
|
||||
});
|
||||
throw rollbackError;
|
||||
try {
|
||||
const rolledBack = await transactionCommand("rollback");
|
||||
if (!workerWorkspaceCommandSucceeded(rolledBack)) {
|
||||
throw workspaceSyncError(rolledBack);
|
||||
}
|
||||
} catch (rollbackFailure) {
|
||||
throw acceptedWorkspaceRollbackError(error, rollbackFailure);
|
||||
}
|
||||
}
|
||||
throw error;
|
||||
|
||||
@@ -136,362 +136,3 @@ function resolveManifest(manifestRoot, requestedDigest) {
|
||||
}
|
||||
fail("worker workspace manifest is unavailable: " + requestedDigest);
|
||||
}`;
|
||||
|
||||
export const REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS = String.raw`const crypto = require("node:crypto");
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const action = process.argv[1];
|
||||
const root = fs.realpathSync(process.argv[2]);
|
||||
const nonce = process.argv[3];
|
||||
if (!/^[a-f0-9]{32}$/.test(nonce || "")) throw new Error("invalid accepted workspace transaction");
|
||||
// REMOTE_WORKSPACE_SETUP_SCRIPT creates and chmods every workspace parent for this worker.
|
||||
// Keeping the transaction beside the workspace makes all live swaps same-filesystem renames.
|
||||
const transactionRoot = path.dirname(root);
|
||||
const transactionRootStats = fs.lstatSync(transactionRoot);
|
||||
if (transactionRootStats.isSymbolicLink() || !transactionRootStats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace transaction directory");
|
||||
}
|
||||
const workspaceKey = crypto.createHash("sha256").update(root).digest("hex");
|
||||
const transactionPrefix = ".openclaw-accepted-" + workspaceKey + "-";
|
||||
const cleanupPrefix = ".openclaw-accepted-cleanup-" + workspaceKey + "-";
|
||||
const transaction = path.join(transactionRoot, transactionPrefix + nonce);
|
||||
const cleanup = path.join(transactionRoot, cleanupPrefix + nonce);
|
||||
const nextRoot = path.join(transaction, "next");
|
||||
const backupRoot = path.join(transaction, "backup");
|
||||
const pathsFile = path.join(transaction, "paths.json");
|
||||
const stateFile = path.join(transaction, "state.json");
|
||||
const ancestorModesFile = path.join(transaction, "ancestor-modes.json");
|
||||
const appliedFile = path.join(transaction, "applied");
|
||||
function isSafeRelativePath(relative) {
|
||||
return (
|
||||
typeof relative === "string" &&
|
||||
relative &&
|
||||
!relative.includes("\\") &&
|
||||
!path.posix.isAbsolute(relative) &&
|
||||
path.posix.normalize(relative) === relative &&
|
||||
relative !== "." &&
|
||||
relative !== ".." &&
|
||||
relative !== ".git" &&
|
||||
!relative.startsWith(".git/") &&
|
||||
!relative.startsWith("../")
|
||||
);
|
||||
}
|
||||
function parsePaths(raw) {
|
||||
const values = JSON.parse(raw);
|
||||
if (!Array.isArray(values) || values.length > 25_000) {
|
||||
throw new Error("invalid accepted workspace paths");
|
||||
}
|
||||
const paths = [...new Set(values)];
|
||||
for (const relative of paths) {
|
||||
if (!isSafeRelativePath(relative)) {
|
||||
throw new Error("unsafe accepted workspace path");
|
||||
}
|
||||
}
|
||||
const selected = new Set(paths);
|
||||
// Directory modes are canonical, so a changed directory is added, removed, or
|
||||
// replaced and all of its accepted descendants are changed and staged too.
|
||||
return paths
|
||||
.filter((relative) => {
|
||||
const segments = relative.split("/");
|
||||
for (let index = 1; index < segments.length; index += 1) {
|
||||
if (selected.has(segments.slice(0, index).join("/"))) return false;
|
||||
}
|
||||
return true;
|
||||
})
|
||||
.sort();
|
||||
}
|
||||
function targetPath(base, relative) {
|
||||
return path.join(base, relative);
|
||||
}
|
||||
function livePath(relative) {
|
||||
const segments = relative.split("/");
|
||||
let parent = root;
|
||||
for (const segment of segments.slice(0, -1)) {
|
||||
parent = path.join(parent, segment);
|
||||
const stats = fs.lstatSync(parent);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
}
|
||||
return path.join(root, relative);
|
||||
}
|
||||
function exists(target) {
|
||||
try {
|
||||
fs.lstatSync(target);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return false;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
function removeTree(target) {
|
||||
let stats;
|
||||
try {
|
||||
stats = fs.lstatSync(target);
|
||||
} catch (error) {
|
||||
if (error && error.code === "ENOENT") return;
|
||||
throw error;
|
||||
}
|
||||
if (stats.isDirectory() && !stats.isSymbolicLink()) {
|
||||
fs.chmodSync(target, 0o700);
|
||||
for (const name of fs.readdirSync(target)) {
|
||||
removeTree(path.join(target, name));
|
||||
}
|
||||
fs.rmdirSync(target);
|
||||
} else {
|
||||
fs.unlinkSync(target);
|
||||
}
|
||||
}
|
||||
function readPaths() {
|
||||
return parsePaths(fs.readFileSync(pathsFile, "utf8"));
|
||||
}
|
||||
function readState(candidate) {
|
||||
const value = JSON.parse(fs.readFileSync(path.join(candidate, "state.json"), "utf8"));
|
||||
if (!Array.isArray(value) || value.length > 25_000) {
|
||||
throw new Error("invalid accepted workspace transaction state");
|
||||
}
|
||||
const relatives = parsePaths(JSON.stringify(value.map((entry) => entry && entry.relative)));
|
||||
if (
|
||||
relatives.length !== value.length ||
|
||||
value.some(
|
||||
(entry, index) =>
|
||||
!entry ||
|
||||
entry.relative !== relatives[index] ||
|
||||
typeof entry.hadLive !== "boolean" ||
|
||||
(entry.directoryMode !== undefined &&
|
||||
(!Number.isInteger(entry.directoryMode) ||
|
||||
entry.directoryMode < 0 ||
|
||||
entry.directoryMode > 0o7777)),
|
||||
)
|
||||
) {
|
||||
throw new Error("invalid accepted workspace transaction state");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
function readAncestorModes(candidate) {
|
||||
const candidateModes = path.join(candidate, "ancestor-modes.json");
|
||||
if (!exists(candidateModes)) return [];
|
||||
const value = JSON.parse(fs.readFileSync(candidateModes, "utf8"));
|
||||
if (!Array.isArray(value) || value.length > 250_000) {
|
||||
throw new Error("invalid accepted workspace ancestor modes");
|
||||
}
|
||||
const seen = new Set();
|
||||
for (const entry of value) {
|
||||
if (
|
||||
!entry ||
|
||||
(entry.relative !== "" && !isSafeRelativePath(entry.relative)) ||
|
||||
seen.has(entry.relative) ||
|
||||
!Number.isInteger(entry.mode) ||
|
||||
entry.mode < 0 ||
|
||||
entry.mode > 0o7777
|
||||
) {
|
||||
throw new Error("invalid accepted workspace ancestor modes");
|
||||
}
|
||||
seen.add(entry.relative);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
function writeAncestorModes(value) {
|
||||
const temporary = ancestorModesFile + ".tmp";
|
||||
fs.writeFileSync(temporary, JSON.stringify(value), { flag: "wx", mode: 0o600 });
|
||||
fs.renameSync(temporary, ancestorModesFile);
|
||||
}
|
||||
function ancestorPaths(paths) {
|
||||
const ancestors = new Set();
|
||||
for (const relative of paths) {
|
||||
const segments = relative.split("/");
|
||||
for (let index = 1; index < segments.length; index += 1) {
|
||||
ancestors.add(segments.slice(0, index).join("/"));
|
||||
}
|
||||
}
|
||||
if (ancestors.size + 1 > 250_000) {
|
||||
throw new Error("accepted workspace transaction has too many ancestors");
|
||||
}
|
||||
return [...ancestors].sort((left, right) => {
|
||||
const depth = left.split("/").length - right.split("/").length;
|
||||
return depth || (left < right ? -1 : left > right ? 1 : 0);
|
||||
});
|
||||
}
|
||||
function prepareWritableAncestors(paths) {
|
||||
// parsePaths removes descendants of changed directories, so these are all
|
||||
// unchanged live ancestors. Read every mode before mutating any permission.
|
||||
const modes = ["", ...ancestorPaths(paths)].map((relative) => {
|
||||
const target = relative ? targetPath(root, relative) : root;
|
||||
const stats = fs.lstatSync(target);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
return { relative, mode: stats.mode & 0o7777 };
|
||||
});
|
||||
writeAncestorModes(modes);
|
||||
makeAncestorsWritable(modes);
|
||||
return modes;
|
||||
}
|
||||
function makeAncestorsWritable(modes) {
|
||||
const widened = [];
|
||||
try {
|
||||
for (const entry of modes) {
|
||||
const target = entry.relative ? targetPath(root, entry.relative) : root;
|
||||
const stats = fs.lstatSync(target);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
const currentMode = stats.mode & 0o7777;
|
||||
const writableMode = entry.mode | 0o700;
|
||||
if (currentMode !== writableMode) {
|
||||
fs.chmodSync(target, writableMode);
|
||||
widened.push(entry);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
try {
|
||||
restoreAncestorModes(widened);
|
||||
} catch (restoreError) {
|
||||
const failure = new Error("accepted workspace ancestor mode rollback failed", {
|
||||
cause: error,
|
||||
});
|
||||
Object.defineProperty(failure, "restoreFailure", { value: restoreError });
|
||||
throw failure;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
function restoreAncestorModes(modes) {
|
||||
for (const entry of [...modes].reverse()) {
|
||||
const target = entry.relative ? targetPath(root, entry.relative) : root;
|
||||
const stats = fs.lstatSync(target);
|
||||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||||
throw new Error("unsafe accepted workspace parent");
|
||||
}
|
||||
if ((stats.mode & 0o7777) !== entry.mode) {
|
||||
fs.chmodSync(target, entry.mode);
|
||||
}
|
||||
}
|
||||
}
|
||||
function removeTransaction(candidate = transaction) {
|
||||
removeTree(candidate);
|
||||
}
|
||||
function restoreTransaction(candidate) {
|
||||
if (!exists(candidate)) return;
|
||||
const ancestorModes = readAncestorModes(candidate);
|
||||
makeAncestorsWritable(ancestorModes);
|
||||
const candidateState = path.join(candidate, "state.json");
|
||||
try {
|
||||
if (exists(candidateState)) {
|
||||
const candidateBackup = path.join(candidate, "backup");
|
||||
for (const entry of [...readState(candidate)].reverse()) {
|
||||
const live = livePath(entry.relative);
|
||||
const backup = targetPath(candidateBackup, entry.relative);
|
||||
if (exists(backup)) {
|
||||
removeTree(live);
|
||||
fs.renameSync(backup, live);
|
||||
if (entry.directoryMode !== undefined) {
|
||||
fs.chmodSync(live, entry.directoryMode);
|
||||
}
|
||||
} else if (!entry.hadLive) {
|
||||
removeTree(live);
|
||||
} else if (entry.directoryMode !== undefined && exists(live)) {
|
||||
fs.chmodSync(live, entry.directoryMode);
|
||||
}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
restoreAncestorModes(ancestorModes);
|
||||
}
|
||||
removeTransaction(candidate);
|
||||
}
|
||||
function recoverTransaction(candidate) {
|
||||
restoreTransaction(candidate);
|
||||
}
|
||||
function recoverTransactions() {
|
||||
for (const name of fs.readdirSync(transactionRoot)) {
|
||||
if (
|
||||
name.startsWith(cleanupPrefix) &&
|
||||
/^[a-f0-9]{32}$/.test(name.slice(cleanupPrefix.length))
|
||||
) {
|
||||
removeTransaction(path.join(transactionRoot, name));
|
||||
}
|
||||
}
|
||||
for (const name of fs.readdirSync(transactionRoot)) {
|
||||
if (
|
||||
name.startsWith(transactionPrefix) &&
|
||||
/^[a-f0-9]{32}$/.test(name.slice(transactionPrefix.length))
|
||||
) {
|
||||
recoverTransaction(path.join(transactionRoot, name));
|
||||
}
|
||||
}
|
||||
}
|
||||
if (action === "begin") {
|
||||
const paths = parsePaths(fs.readFileSync(0, "utf8"));
|
||||
recoverTransactions();
|
||||
fs.mkdirSync(transaction, { mode: 0o700 });
|
||||
fs.mkdirSync(nextRoot, { mode: 0o700 });
|
||||
fs.mkdirSync(backupRoot, { mode: 0o700 });
|
||||
fs.writeFileSync(pathsFile, JSON.stringify(paths), { mode: 0o600 });
|
||||
process.stdout.write(nextRoot + "\n");
|
||||
} else if (action === "apply") {
|
||||
const paths = readPaths();
|
||||
try {
|
||||
const ancestorModes = prepareWritableAncestors(paths);
|
||||
const state = paths.map((relative) => {
|
||||
const live = livePath(relative);
|
||||
if (!exists(live)) return { relative, hadLive: false };
|
||||
const stats = fs.lstatSync(live);
|
||||
return {
|
||||
relative,
|
||||
hadLive: true,
|
||||
...(stats.isDirectory() && !stats.isSymbolicLink()
|
||||
? { directoryMode: stats.mode & 0o7777 }
|
||||
: {}),
|
||||
};
|
||||
});
|
||||
const temporaryStateFile = stateFile + ".tmp";
|
||||
fs.writeFileSync(temporaryStateFile, JSON.stringify(state), { flag: "wx", mode: 0o600 });
|
||||
fs.renameSync(temporaryStateFile, stateFile);
|
||||
for (const entry of state) {
|
||||
if (!entry.hadLive) continue;
|
||||
const source = livePath(entry.relative);
|
||||
const sourceStats = fs.lstatSync(source);
|
||||
const destination = targetPath(backupRoot, entry.relative);
|
||||
fs.mkdirSync(path.dirname(destination), { recursive: true, mode: 0o700 });
|
||||
try {
|
||||
if (sourceStats.isDirectory() && !sourceStats.isSymbolicLink()) {
|
||||
fs.chmodSync(source, 0o700);
|
||||
}
|
||||
fs.renameSync(source, destination);
|
||||
} catch (error) {
|
||||
if (entry.directoryMode !== undefined && exists(source)) {
|
||||
fs.chmodSync(source, entry.directoryMode);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
for (const entry of state) {
|
||||
const source = targetPath(nextRoot, entry.relative);
|
||||
if (!exists(source)) continue;
|
||||
fs.renameSync(source, livePath(entry.relative));
|
||||
}
|
||||
restoreAncestorModes(ancestorModes);
|
||||
fs.writeFileSync(appliedFile, "", { flag: "wx", mode: 0o600 });
|
||||
} catch (error) {
|
||||
restoreTransaction(transaction);
|
||||
throw error;
|
||||
}
|
||||
} else if (action === "rollback") {
|
||||
if (exists(cleanup)) {
|
||||
if (exists(transaction)) throw new Error("ambiguous accepted workspace transaction state");
|
||||
fs.renameSync(cleanup, transaction);
|
||||
}
|
||||
restoreTransaction(transaction);
|
||||
} else if (action === "recover") {
|
||||
recoverTransactions();
|
||||
} else if (action === "commit") {
|
||||
if (exists(transaction)) {
|
||||
if (!exists(appliedFile)) throw new Error("accepted workspace transaction is not applied");
|
||||
// The namespace rename is the commit point. Later recovery removes the backup
|
||||
// only after the gateway has had a chance to observe this command's success.
|
||||
fs.renameSync(transaction, cleanup);
|
||||
}
|
||||
} else {
|
||||
throw new Error("invalid accepted workspace transaction action");
|
||||
}`;
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { createHash, randomBytes } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import { isAcceptedWorkspacePublicationIndeterminateError } from "./workspace-accepted-publication.js";
|
||||
import {
|
||||
MAX_RECONCILIATION_ENTRIES,
|
||||
type WorkerWorkspaceManifest,
|
||||
@@ -209,6 +210,11 @@ export async function applyStagedWorkerWorkspace(params: {
|
||||
}),
|
||||
};
|
||||
} catch (error) {
|
||||
// Transport or settlement timeouts are observation evidence, never authority
|
||||
// for an inverse operation; recovery owns restoring both sides.
|
||||
if (isAcceptedWorkspacePublicationIndeterminateError(error)) {
|
||||
throw error;
|
||||
}
|
||||
try {
|
||||
await recoverWorkerWorkspaceReconciliation({ root, journal });
|
||||
params.journal.abort();
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
|
||||
import { AcceptedWorkspacePublicationIndeterminateError } from "./workspace-accepted-publication.js";
|
||||
import {
|
||||
applyStagedWorkerWorkspace,
|
||||
readActualWorkspaceManifest,
|
||||
recoverWorkerWorkspaceReconciliation,
|
||||
type WorkerWorkspaceReconciliationJournal,
|
||||
} from "./workspace-reconcile.js";
|
||||
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
async function temporaryDirectory(name: string): Promise<string> {
|
||||
return await fs.realpath(tempDirs.make(`openclaw-${name}-`));
|
||||
}
|
||||
|
||||
async function manifestFor(root: string) {
|
||||
return (await readActualWorkspaceManifest({ root, baseCommit: null })).manifest;
|
||||
}
|
||||
|
||||
describe("worker workspace reconciliation publication", () => {
|
||||
it("keeps local bytes and the journal pending when accepted publication is indeterminate", async () => {
|
||||
const local = await temporaryDirectory("workspace-indeterminate-publication");
|
||||
const staged = await temporaryDirectory("workspace-indeterminate-publication-staged");
|
||||
await fs.writeFile(path.join(local, "result.txt"), "base\n");
|
||||
const base = await manifestFor(local);
|
||||
await Promise.all([
|
||||
fs.writeFile(path.join(staged, "result.txt"), "worker\n"),
|
||||
fs.writeFile(path.join(staged, "added.txt"), "added\n"),
|
||||
]);
|
||||
const current = await manifestFor(staged);
|
||||
let pending: WorkerWorkspaceReconciliationJournal | undefined;
|
||||
const abort = vi.fn(() => {
|
||||
pending = undefined;
|
||||
});
|
||||
const commit = vi.fn(() => {
|
||||
pending = undefined;
|
||||
});
|
||||
const journal = {
|
||||
load: () => pending,
|
||||
begin: (value: WorkerWorkspaceReconciliationJournal) => {
|
||||
pending = value;
|
||||
},
|
||||
commit,
|
||||
abort,
|
||||
};
|
||||
const publicationFailure = new AcceptedWorkspacePublicationIndeterminateError(
|
||||
"apply",
|
||||
new Error("apply transport lost"),
|
||||
new Error("settlement timed out"),
|
||||
);
|
||||
|
||||
await expect(
|
||||
applyStagedWorkerWorkspace({
|
||||
root: local,
|
||||
stagingRoot: staged,
|
||||
baseManifestRef: `sha256:${"a".repeat(64)}`,
|
||||
currentManifestRef: `sha256:${"b".repeat(64)}`,
|
||||
base,
|
||||
current,
|
||||
journal,
|
||||
publishAcceptedManifest: async () => {
|
||||
throw publicationFailure;
|
||||
},
|
||||
}),
|
||||
).rejects.toBe(publicationFailure);
|
||||
|
||||
expect(pending).toBeDefined();
|
||||
expect(commit).not.toHaveBeenCalled();
|
||||
expect(abort).not.toHaveBeenCalled();
|
||||
await expect(fs.readFile(path.join(local, "result.txt"), "utf8")).resolves.toBe("worker\n");
|
||||
await expect(fs.readFile(path.join(local, "added.txt"), "utf8")).resolves.toBe("added\n");
|
||||
|
||||
await recoverWorkerWorkspaceReconciliation({ root: local, journal: pending! });
|
||||
await expect(fs.readFile(path.join(local, "result.txt"), "utf8")).resolves.toBe("base\n");
|
||||
await expect(fs.access(path.join(local, "added.txt"))).rejects.toThrow();
|
||||
expect(pending).toBeDefined();
|
||||
journal.abort();
|
||||
expect(pending).toBeUndefined();
|
||||
expect(abort).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("rolls local bytes back immediately when accepted publication fails definitively", async () => {
|
||||
const local = await temporaryDirectory("workspace-definitive-publication-failure");
|
||||
const staged = await temporaryDirectory("workspace-definitive-publication-failure-staged");
|
||||
await fs.writeFile(path.join(local, "result.txt"), "base\n");
|
||||
const base = await manifestFor(local);
|
||||
await fs.writeFile(path.join(staged, "result.txt"), "worker\n");
|
||||
const current = await manifestFor(staged);
|
||||
let pending: WorkerWorkspaceReconciliationJournal | undefined;
|
||||
const abort = vi.fn(() => {
|
||||
pending = undefined;
|
||||
});
|
||||
|
||||
await expect(
|
||||
applyStagedWorkerWorkspace({
|
||||
root: local,
|
||||
stagingRoot: staged,
|
||||
baseManifestRef: `sha256:${"a".repeat(64)}`,
|
||||
currentManifestRef: `sha256:${"b".repeat(64)}`,
|
||||
base,
|
||||
current,
|
||||
journal: {
|
||||
load: () => pending,
|
||||
begin: (value) => {
|
||||
pending = value;
|
||||
},
|
||||
commit: () => {
|
||||
pending = undefined;
|
||||
},
|
||||
abort,
|
||||
},
|
||||
publishAcceptedManifest: async () => {
|
||||
throw new Error("publication rejected");
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("publication rejected");
|
||||
|
||||
await expect(fs.readFile(path.join(local, "result.txt"), "utf8")).resolves.toBe("base\n");
|
||||
expect(pending).toBeUndefined();
|
||||
expect(abort).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
@@ -2,9 +2,10 @@ import { spawn } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import { once } from "node:events";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { waitForChildClose, waitForFile } from "../../../test/helpers/process-wait.js";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
|
||||
import { runCommandWithTimeout } from "../../process/exec.js";
|
||||
import {
|
||||
parseWorkerWorkspaceManifest,
|
||||
@@ -20,15 +21,25 @@ import {
|
||||
REMOTE_WORKSPACE_MANIFEST_JS,
|
||||
} from "./workspace-sync-scripts.js";
|
||||
|
||||
const roots: string[] = [];
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));
|
||||
});
|
||||
function spawnTransaction(argv: string[], env: NodeJS.ProcessEnv) {
|
||||
const child = spawn(process.execPath, argv, { env, stdio: ["ignore", "pipe", "pipe"] });
|
||||
let stderr = "";
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stderr.on("data", (chunk: string) => {
|
||||
stderr += chunk;
|
||||
});
|
||||
const exited = waitForChildClose(child, 10_000).then(({ code, signal }) => ({
|
||||
code,
|
||||
signal,
|
||||
stderr,
|
||||
}));
|
||||
return { exited };
|
||||
}
|
||||
|
||||
async function fixture() {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-quiescence-test-"));
|
||||
roots.push(root);
|
||||
const root = tempDirs.make("openclaw-quiescence-test-");
|
||||
const home = path.join(root, "home");
|
||||
let workspace = path.join(root, "workspace");
|
||||
const bin = path.join(root, "bin");
|
||||
@@ -207,8 +218,7 @@ describe("remote workspace quiescence scripts", () => {
|
||||
|
||||
describe("remote workspace manifest script", () => {
|
||||
it("atomically applies and rolls back accepted workspace paths", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-accepted-paths-test-"));
|
||||
roots.push(root);
|
||||
const root = tempDirs.make("openclaw-accepted-paths-test-");
|
||||
const home = path.join(root, "home");
|
||||
const workspace = path.join(root, "workspace");
|
||||
await Promise.all([fs.mkdir(home), fs.mkdir(workspace)]);
|
||||
@@ -251,13 +261,31 @@ describe("remote workspace manifest script", () => {
|
||||
);
|
||||
await expect(fs.readFile(path.join(workspace, "added.txt"), "utf8")).resolves.toBe("added\n");
|
||||
|
||||
await fs.rm(path.join(path.dirname(staging), "applied"));
|
||||
await fs.writeFile(
|
||||
path.join(path.dirname(staging), "phase.json"),
|
||||
JSON.stringify({ version: 1, nonce, phase: "applying" }),
|
||||
);
|
||||
const recoveryNonce = "b".repeat(32);
|
||||
const recoveryBegin = await runTransaction("begin", recoveryNonce, JSON.stringify(["node"]));
|
||||
expect(recoveryBegin.code).toBe(0);
|
||||
await expect(fs.readFile(path.join(workspace, "node"), "utf8")).resolves.toBe("old file\n");
|
||||
await expect(fs.access(path.join(workspace, "added.txt"))).rejects.toThrow();
|
||||
expect((await runTransaction("rollback", recoveryNonce)).code).toBe(0);
|
||||
|
||||
const legacyNonce = "6".repeat(32);
|
||||
const legacyBegin = await runTransaction("begin", legacyNonce, JSON.stringify(["node"]));
|
||||
const legacyTransaction = path.dirname(legacyBegin.stdout.trim());
|
||||
await fs.writeFile(path.join(legacyBegin.stdout.trim(), "node"), "legacy applied\n");
|
||||
expect((await runTransaction("apply", legacyNonce)).code).toBe(0);
|
||||
await fs.rm(path.join(legacyTransaction, "phase.json"));
|
||||
await fs.writeFile(path.join(legacyTransaction, "applied"), "");
|
||||
const legacyRecoveryNonce = "7".repeat(32);
|
||||
expect(
|
||||
await runTransaction("begin", legacyRecoveryNonce, JSON.stringify(["node"])),
|
||||
).toMatchObject({ code: 0 });
|
||||
await expect(fs.readFile(path.join(workspace, "node"), "utf8")).resolves.toBe("old file\n");
|
||||
expect((await runTransaction("rollback", legacyRecoveryNonce)).code).toBe(0);
|
||||
|
||||
await fs.rm(path.join(workspace, "node"));
|
||||
await fs.mkdir(path.join(workspace, "node"));
|
||||
await fs.writeFile(path.join(workspace, "node/old.txt"), "read only\n");
|
||||
@@ -355,9 +383,248 @@ describe("remote workspace manifest script", () => {
|
||||
await fs.chmod(path.join(workspace, "parent"), 0o700);
|
||||
});
|
||||
|
||||
it("reports strict settlement outcomes for each durable transaction phase", async () => {
|
||||
const root = tempDirs.make("openclaw-accepted-settlement-outcomes-");
|
||||
let workspace = path.join(root, "workspace");
|
||||
await fs.mkdir(workspace);
|
||||
workspace = await fs.realpath(workspace);
|
||||
await fs.writeFile(path.join(workspace, "result.txt"), "old\n");
|
||||
const runTransaction = async (action: string, nonce: string, input?: string) =>
|
||||
await runCommandWithTimeout(
|
||||
[
|
||||
process.execPath,
|
||||
"-e",
|
||||
REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS,
|
||||
action,
|
||||
workspace,
|
||||
nonce,
|
||||
],
|
||||
{ timeoutMs: 10_000, input },
|
||||
);
|
||||
const expectSettlement = (
|
||||
value: Awaited<ReturnType<typeof runTransaction>>,
|
||||
outcome: "begun" | "rolled-back" | "applied" | "committed",
|
||||
) => {
|
||||
expect(value).toMatchObject({
|
||||
code: 0,
|
||||
stderr: "",
|
||||
stdout: `${JSON.stringify({ version: 1, outcome })}\n`,
|
||||
});
|
||||
};
|
||||
|
||||
const begunNonce = "a".repeat(32);
|
||||
expect(await runTransaction("begin", begunNonce, JSON.stringify(["result.txt"]))).toMatchObject(
|
||||
{ code: 0 },
|
||||
);
|
||||
expectSettlement(await runTransaction("settle", begunNonce), "begun");
|
||||
expect(await runTransaction("rollback", begunNonce)).toMatchObject({ code: 0 });
|
||||
|
||||
const appliedNonce = "b".repeat(32);
|
||||
const appliedBegin = await runTransaction(
|
||||
"begin",
|
||||
appliedNonce,
|
||||
JSON.stringify(["result.txt"]),
|
||||
);
|
||||
await fs.writeFile(path.join(appliedBegin.stdout.trim(), "result.txt"), "applied\n");
|
||||
expect(await runTransaction("apply", appliedNonce)).toMatchObject({ code: 0 });
|
||||
expectSettlement(await runTransaction("settle", appliedNonce), "applied");
|
||||
expect(await runTransaction("rollback", appliedNonce)).toMatchObject({ code: 0 });
|
||||
|
||||
const committedNonce = "c".repeat(32);
|
||||
const committedBegin = await runTransaction(
|
||||
"begin",
|
||||
committedNonce,
|
||||
JSON.stringify(["result.txt"]),
|
||||
);
|
||||
await fs.writeFile(path.join(committedBegin.stdout.trim(), "result.txt"), "committed\n");
|
||||
expect(await runTransaction("apply", committedNonce)).toMatchObject({ code: 0 });
|
||||
expect(await runTransaction("commit", committedNonce)).toMatchObject({ code: 0 });
|
||||
expectSettlement(await runTransaction("settle", committedNonce), "committed");
|
||||
expect(await runTransaction("recover", "d".repeat(32))).toMatchObject({ code: 0 });
|
||||
await expect(fs.readFile(path.join(workspace, "result.txt"), "utf8")).resolves.toBe(
|
||||
"committed\n",
|
||||
);
|
||||
expect(
|
||||
(await fs.readdir(root)).filter((name) => name.startsWith(".openclaw-accepted-")),
|
||||
).toEqual([]);
|
||||
});
|
||||
|
||||
it("serializes a live apply against rollback and recovery", async () => {
|
||||
for (const contender of ["rollback", "recover"] as const) {
|
||||
const root = tempDirs.make(`openclaw-accepted-${contender}-`);
|
||||
let workspace = path.join(root, "workspace");
|
||||
const gate = path.join(root, "gate.fifo");
|
||||
const applyMarker = path.join(root, "apply-started");
|
||||
const contenderMarker = path.join(root, "contender-waiting");
|
||||
const preload = path.join(root, "gate.cjs");
|
||||
await fs.mkdir(workspace);
|
||||
workspace = await fs.realpath(workspace);
|
||||
await fs.writeFile(path.join(workspace, "result.txt"), "old\n");
|
||||
const mkfifo = await runCommandWithTimeout(["mkfifo", gate], { timeoutMs: 10_000 });
|
||||
expect(mkfifo.code).toBe(0);
|
||||
await fs.writeFile(
|
||||
preload,
|
||||
`const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const renameSync = fs.renameSync;
|
||||
let applyGated = false;
|
||||
fs.renameSync = function(source, destination) {
|
||||
const result = renameSync.apply(this, arguments);
|
||||
if (!applyGated && process.argv[1] === "apply" && source === process.env.OPENCLAW_TEST_GATE_SOURCE && destination.includes(path.sep + "backup" + path.sep)) {
|
||||
applyGated = true;
|
||||
fs.writeFileSync(process.env.OPENCLAW_TEST_APPLY_MARKER, "");
|
||||
fs.readFileSync(process.env.OPENCLAW_TEST_GATE);
|
||||
}
|
||||
return result;
|
||||
};
|
||||
const kill = process.kill.bind(process);
|
||||
let contenderMarked = false;
|
||||
process.kill = function(pid, signal) {
|
||||
if (!contenderMarked && signal === 0 && process.argv[1] === process.env.OPENCLAW_TEST_CONTENDER) {
|
||||
contenderMarked = true;
|
||||
fs.writeFileSync(process.env.OPENCLAW_TEST_CONTENDER_MARKER, "");
|
||||
}
|
||||
return kill(pid, signal);
|
||||
};
|
||||
`,
|
||||
);
|
||||
const env = {
|
||||
...process.env,
|
||||
OPENCLAW_TEST_GATE: gate,
|
||||
OPENCLAW_TEST_GATE_SOURCE: path.join(workspace, "result.txt"),
|
||||
OPENCLAW_TEST_APPLY_MARKER: applyMarker,
|
||||
OPENCLAW_TEST_CONTENDER: contender,
|
||||
OPENCLAW_TEST_CONTENDER_MARKER: contenderMarker,
|
||||
};
|
||||
const nonce = contender === "rollback" ? "3".repeat(32) : "4".repeat(32);
|
||||
const begin = await runCommandWithTimeout(
|
||||
[
|
||||
process.execPath,
|
||||
"-e",
|
||||
REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS,
|
||||
"begin",
|
||||
workspace,
|
||||
nonce,
|
||||
],
|
||||
{ timeoutMs: 10_000, baseEnv: env, input: JSON.stringify(["result.txt"]) },
|
||||
);
|
||||
expect(begin.code).toBe(0);
|
||||
const staging = begin.stdout.trim();
|
||||
await fs.writeFile(path.join(staging, "result.txt"), "new\n");
|
||||
|
||||
const apply = spawnTransaction(
|
||||
[
|
||||
"--require",
|
||||
preload,
|
||||
"-e",
|
||||
REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS,
|
||||
"apply",
|
||||
workspace,
|
||||
nonce,
|
||||
],
|
||||
env,
|
||||
);
|
||||
await waitForFile(applyMarker, 10_000);
|
||||
const transaction = path.dirname(staging);
|
||||
await expect(fs.access(path.join(workspace, "result.txt"))).rejects.toThrow();
|
||||
await expect(fs.readFile(path.join(transaction, "backup/result.txt"), "utf8")).resolves.toBe(
|
||||
"old\n",
|
||||
);
|
||||
await expect(fs.readFile(path.join(staging, "result.txt"), "utf8")).resolves.toBe("new\n");
|
||||
|
||||
const contenderNonce = contender === "rollback" ? nonce : "5".repeat(32);
|
||||
const competing = runCommandWithTimeout(
|
||||
[
|
||||
process.execPath,
|
||||
"--require",
|
||||
preload,
|
||||
"-e",
|
||||
REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS,
|
||||
contender,
|
||||
workspace,
|
||||
contenderNonce,
|
||||
],
|
||||
{ timeoutMs: 10_000, baseEnv: env },
|
||||
);
|
||||
await waitForFile(contenderMarker, 10_000);
|
||||
await expect(fs.access(path.join(workspace, "result.txt"))).rejects.toThrow();
|
||||
await expect(fs.readFile(path.join(transaction, "backup/result.txt"), "utf8")).resolves.toBe(
|
||||
"old\n",
|
||||
);
|
||||
|
||||
const gateWriter = await fs.open(gate, "w");
|
||||
await gateWriter.write("release");
|
||||
await gateWriter.close();
|
||||
expect(await apply.exited).toMatchObject({ code: 0, signal: null, stderr: "" });
|
||||
expect(await competing).toMatchObject({ code: 0, stderr: "" });
|
||||
await expect(fs.readFile(path.join(workspace, "result.txt"), "utf8")).resolves.toBe("old\n");
|
||||
expect(
|
||||
(await fs.readdir(root)).filter((name) => name.startsWith(".openclaw-accepted-")),
|
||||
).toEqual([]);
|
||||
}
|
||||
});
|
||||
|
||||
it("restores a dead reclaimer before settling its dead apply owner", async () => {
|
||||
const root = tempDirs.make("openclaw-accepted-dead-owner-");
|
||||
let workspace = path.join(root, "workspace");
|
||||
await fs.mkdir(workspace);
|
||||
workspace = await fs.realpath(workspace);
|
||||
await fs.writeFile(path.join(workspace, "result.txt"), "old\n");
|
||||
const nonce = "8".repeat(32);
|
||||
const runTransaction = async (action: string, input?: string) =>
|
||||
await runCommandWithTimeout(
|
||||
[
|
||||
process.execPath,
|
||||
"-e",
|
||||
REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS,
|
||||
action,
|
||||
workspace,
|
||||
nonce,
|
||||
],
|
||||
{ timeoutMs: 10_000, input },
|
||||
);
|
||||
const begin = await runTransaction("begin", JSON.stringify(["result.txt"]));
|
||||
expect(begin.code).toBe(0);
|
||||
const transaction = path.dirname(begin.stdout.trim());
|
||||
await Promise.all([
|
||||
fs.writeFile(
|
||||
path.join(transaction, "phase.json"),
|
||||
JSON.stringify({ version: 1, nonce, phase: "applying" }),
|
||||
),
|
||||
fs.writeFile(
|
||||
path.join(transaction, "state.json"),
|
||||
JSON.stringify([{ relative: "result.txt", hadLive: true }]),
|
||||
),
|
||||
]);
|
||||
await fs.rename(
|
||||
path.join(workspace, "result.txt"),
|
||||
path.join(transaction, "backup/result.txt"),
|
||||
);
|
||||
const workspaceKey = createHash("sha256").update(workspace).digest("hex");
|
||||
const lock = path.join(root, `.openclaw-accepted-lock-${workspaceKey}`);
|
||||
const deadPid = 2_147_483_647;
|
||||
const token = "9".repeat(32);
|
||||
await fs.mkdir(lock);
|
||||
const ownerIdentity = ["apply", nonce, deadPid, token].join(".");
|
||||
const reclaimToken = "a".repeat(32);
|
||||
const reclaimerIdentity = ["settle", nonce, deadPid, reclaimToken].join(".");
|
||||
await fs.writeFile(path.join(lock, `reclaim.${ownerIdentity}.${reclaimerIdentity}`), "");
|
||||
|
||||
const settled = await runTransaction("settle");
|
||||
|
||||
expect(settled).toMatchObject({
|
||||
code: 0,
|
||||
stderr: "",
|
||||
stdout: `${JSON.stringify({ version: 1, outcome: "rolled-back" })}\n`,
|
||||
});
|
||||
await expect(fs.readFile(path.join(workspace, "result.txt"), "utf8")).resolves.toBe("old\n");
|
||||
expect(
|
||||
(await fs.readdir(root)).filter((name) => name.startsWith(".openclaw-accepted-")),
|
||||
).toEqual([]);
|
||||
});
|
||||
|
||||
it("keeps the gateway's canonical manifest available across a second turn", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-manifest-lifecycle-test-"));
|
||||
roots.push(root);
|
||||
const root = tempDirs.make("openclaw-manifest-lifecycle-test-");
|
||||
const home = path.join(root, "home");
|
||||
const workspace = path.join(root, "workspace");
|
||||
await Promise.all([fs.mkdir(home), fs.mkdir(workspace)]);
|
||||
@@ -511,8 +778,7 @@ describe("remote workspace manifest script", () => {
|
||||
});
|
||||
|
||||
it("drops derived artifacts from the worker manifest", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-manifest-derived-test-"));
|
||||
roots.push(root);
|
||||
const root = tempDirs.make("openclaw-manifest-derived-test-");
|
||||
const home = path.join(root, "home");
|
||||
const workspace = path.join(root, "workspace");
|
||||
const files = [
|
||||
@@ -553,8 +819,7 @@ describe("remote workspace manifest script", () => {
|
||||
});
|
||||
|
||||
it("keeps base tombstones in the final ignored-path verification", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-manifest-tombstone-test-"));
|
||||
roots.push(root);
|
||||
const root = tempDirs.make("openclaw-manifest-tombstone-test-");
|
||||
const home = path.join(root, "home");
|
||||
const workspace = path.join(root, "workspace");
|
||||
await fs.mkdir(home);
|
||||
@@ -632,8 +897,7 @@ describe("remote workspace manifest script", () => {
|
||||
});
|
||||
|
||||
it("drops stale descendants when a tracked directory becomes a file", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-manifest-test-"));
|
||||
roots.push(root);
|
||||
const root = tempDirs.make("openclaw-manifest-test-");
|
||||
const home = path.join(root, "home");
|
||||
const workspace = path.join(root, "workspace");
|
||||
await fs.mkdir(home);
|
||||
|
||||
@@ -2,7 +2,7 @@ import {
|
||||
REMOTE_WORKSPACE_MANIFEST_CANONICAL_JS,
|
||||
REMOTE_WORKSPACE_MANIFEST_REGISTRY_JS,
|
||||
} from "./workspace-manifest-remote-script.js";
|
||||
export { REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS } from "./workspace-manifest-remote-script.js";
|
||||
export { REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS } from "./workspace-accepted-remote-script.js";
|
||||
import {
|
||||
DERIVED_WORKSPACE_DIRECTORY_NAMES,
|
||||
DERIVED_WORKSPACE_FILE_NAMES,
|
||||
|
||||
Reference in New Issue
Block a user