diff --git a/src/gateway/worker-environments/workspace-accepted-lock-remote-script.ts b/src/gateway/worker-environments/workspace-accepted-lock-remote-script.ts new file mode 100644 index 000000000000..846cb434b4ce --- /dev/null +++ b/src/gateway/worker-environments/workspace-accepted-lock-remote-script.ts @@ -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); +}`; diff --git a/src/gateway/worker-environments/workspace-accepted-publication.ts b/src/gateway/worker-environments/workspace-accepted-publication.ts new file mode 100644 index 000000000000..98c3189b820a --- /dev/null +++ b/src/gateway/worker-environments/workspace-accepted-publication.ts @@ -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; +} diff --git a/src/gateway/worker-environments/workspace-accepted-remote-script.ts b/src/gateway/worker-environments/workspace-accepted-remote-script.ts new file mode 100644 index 000000000000..d913767b4de5 --- /dev/null +++ b/src/gateway/worker-environments/workspace-accepted-remote-script.ts @@ -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(); +}`; diff --git a/src/gateway/worker-environments/workspace-accepted-sync.test.ts b/src/gateway/worker-environments/workspace-accepted-sync.test.ts new file mode 100644 index 000000000000..133a18dbdd2e --- /dev/null +++ b/src/gateway/worker-environments/workspace-accepted-sync.test.ts @@ -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 { + 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 = []; + let stagingRoot: string | undefined; + let applyExited: + | Promise<{ code: number | null; signal: NodeJS.Signals | null; stderr: string }> + | undefined; + const runWorkspaceCommand = async (command: WorkerWorkspaceCommand): Promise => { + 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 => { + 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 | undefined; + let applyExited: + | Promise<{ code: number | null; signal: NodeJS.Signals | null; stderr: string }> + | undefined; + const runWorkspaceCommand = async (command: WorkerWorkspaceCommand): Promise => { + 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"]); + }); +}); diff --git a/src/gateway/worker-environments/workspace-accepted-sync.ts b/src/gateway/worker-environments/workspace-accepted-sync.ts index 0fc3bbbea474..5b326094c096 100644 --- a/src/gateway/worker-environments/workspace-accepted-sync.ts +++ b/src/gateway/worker-environments/workspace-accepted-sync.ts @@ -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; 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 => { + 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 => { + 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; diff --git a/src/gateway/worker-environments/workspace-manifest-remote-script.ts b/src/gateway/worker-environments/workspace-manifest-remote-script.ts index 9bb35f1420c4..f9103f46bcf7 100644 --- a/src/gateway/worker-environments/workspace-manifest-remote-script.ts +++ b/src/gateway/worker-environments/workspace-manifest-remote-script.ts @@ -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"); -}`; diff --git a/src/gateway/worker-environments/workspace-reconcile-apply.ts b/src/gateway/worker-environments/workspace-reconcile-apply.ts index 2bc6a4447e0d..5cbfaf13a912 100644 --- a/src/gateway/worker-environments/workspace-reconcile-apply.ts +++ b/src/gateway/worker-environments/workspace-reconcile-apply.ts @@ -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(); diff --git a/src/gateway/worker-environments/workspace-reconcile-publication.test.ts b/src/gateway/worker-environments/workspace-reconcile-publication.test.ts new file mode 100644 index 000000000000..04770ace426b --- /dev/null +++ b/src/gateway/worker-environments/workspace-reconcile-publication.test.ts @@ -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 { + 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(); + }); +}); diff --git a/src/gateway/worker-environments/workspace-sync-scripts.test.ts b/src/gateway/worker-environments/workspace-sync-scripts.test.ts index 195fef190367..3d3513b66765 100644 --- a/src/gateway/worker-environments/workspace-sync-scripts.test.ts +++ b/src/gateway/worker-environments/workspace-sync-scripts.test.ts @@ -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>, + 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); diff --git a/src/gateway/worker-environments/workspace-sync-scripts.ts b/src/gateway/worker-environments/workspace-sync-scripts.ts index c6cd66b7d80f..b050c0f41e8a 100644 --- a/src/gateway/worker-environments/workspace-sync-scripts.ts +++ b/src/gateway/worker-environments/workspace-sync-scripts.ts @@ -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,