diff --git a/.gitignore b/.gitignore index 7a02601..a6c6e5c 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,6 @@ artifacts/ screenshots/ traces/ *.log +.mstack-sync-upstream.lock* +.mstack-sync-upstream/ +**/.mstack-sync-upstream-tx/ diff --git a/README.md b/README.md index 0c8a1a5..bca1ec6 100644 --- a/README.md +++ b/README.md @@ -276,6 +276,14 @@ content still matches the previous manifest. Pass `--apply --force` after reviewing a diff when overwriting or removing a locally changed file is intentional. +Only one `--apply` can run against a target at a time. Writes and removals are +staged and journaled before the target is changed; an ordinary failure rolls +the whole refresh back. If the process is interrupted, the next `--apply` +recovers the unfinished transaction before rebuilding the sync plan. Dry runs +and `check-upstream` refuse to inspect a target while a sync is active or needs +recovery. The transaction prevents persistent partial refreshes, but readers +that ignore the sync lock can still observe files changing during the commit. + `--apply` writes `profiles/upstream-manifest.json`. Its hashes describe the transformed upstream baseline, not package integrity. Files configured with `compareContent: false`, including the adapted agents, may intentionally differ diff --git a/scripts/check-upstream.mjs b/scripts/check-upstream.mjs index ea9ea30..0414be5 100644 --- a/scripts/check-upstream.mjs +++ b/scripts/check-upstream.mjs @@ -3,6 +3,11 @@ import { execFileSync } from "node:child_process"; import { existsSync, lstatSync, readdirSync, readFileSync, realpathSync } from "node:fs"; import { dirname, extname, isAbsolute, join, posix, relative, resolve, sep, win32 } from "node:path"; import { fileURLToPath } from "node:url"; +import { + acquireUpstreamSyncLock, + assertUpstreamSyncTargetReadable, + releaseUpstreamSyncLock, +} from "./sync-upstream-transaction.mjs"; const args = process.argv.slice(2); const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); @@ -23,6 +28,16 @@ if (args.includes("--help")) { } const targetRoot = resolve(valueAfter("--target") ?? repoRoot); +const syncLock = acquireUpstreamSyncLock(targetRoot, { recoverStale: false }); +const releaseReadLockAtExit = () => { + try { + releaseUpstreamSyncLock(syncLock); + } catch { + // A stale read lock is recoverable by the next apply. + } +}; +process.once("exit", releaseReadLockAtExit); +assertUpstreamSyncTargetReadable(targetRoot, syncLock); const upstreamsPath = resolveInside(targetRoot, "profiles/upstreams.json", "upstream profile"); if (!existsSync(upstreamsPath)) throw new Error(`mstack upstream profile not found: ${upstreamsPath}`); const upstreams = JSON.parse(readFileSync(upstreamsPath, "utf8")); @@ -338,3 +353,5 @@ if (!existsSync(manifestPath)) { console.log(`Matched ${expected.length} pstack skills at ${pstack.commit}.`); if (artifactProblems && strict) process.exit(1); +releaseUpstreamSyncLock(syncLock); +process.removeListener("exit", releaseReadLockAtExit); diff --git a/scripts/sync-upstream-transaction.mjs b/scripts/sync-upstream-transaction.mjs new file mode 100644 index 0000000..4574cc7 --- /dev/null +++ b/scripts/sync-upstream-transaction.mjs @@ -0,0 +1,878 @@ +import { + closeSync, + existsSync, + fchmodSync, + fsyncSync, + linkSync, + lstatSync, + mkdirSync, + openSync, + readFileSync, + readdirSync, + realpathSync, + renameSync, + rmdirSync, + statSync, + unlinkSync, + writeFileSync, +} from "node:fs"; +import { createHash, randomUUID } from "node:crypto"; +import { basename, dirname, isAbsolute, posix, relative, resolve, sep } from "node:path"; + +const LOCK_NAME = ".mstack-sync-upstream.lock"; +const STATE_NAME = ".mstack-sync-upstream"; +const SIDECAR_NAME = ".mstack-sync-upstream-tx"; +const JOURNAL_VERSION = 1; +const HASH_PATTERN = /^[0-9a-f]{64}$/; + +function digest(value) { + return createHash("sha256").update(value).digest("hex"); +} + +function fsyncDirectory(path) { + let descriptor; + try { + descriptor = openSync(path, "r"); + fsyncSync(descriptor); + } catch (error) { + const unsupportedOnWindows = process.platform === "win32" && + ["EACCES", "EBADF", "EINVAL", "EISDIR", "ENOTSUP", "EPERM"].includes(error.code); + if (!unsupportedOnWindows) throw error; + } finally { + if (descriptor !== undefined) closeSync(descriptor); + } +} + +function writeDurableFile(path, content, mode = 0o600) { + const descriptor = openSync(path, "wx", mode); + try { + writeFileSync(descriptor, content); + fchmodSync(descriptor, mode); + fsyncSync(descriptor); + } finally { + closeSync(descriptor); + } + fsyncDirectory(dirname(path)); +} + +function createDurableMarker(transactionPath, name, content) { + const temporaryPath = resolve(transactionPath, `${name}.tmp`); + const markerPath = resolve(transactionPath, name); + writeDurableFile(temporaryPath, content); + durableRename(temporaryPath, markerPath); +} + +function durableRename(from, to) { + renameSync(from, to); + fsyncDirectory(dirname(from)); + if (dirname(to) !== dirname(from)) fsyncDirectory(dirname(to)); +} + +function durableUnlink(path) { + unlinkSync(path); + fsyncDirectory(dirname(path)); +} + +function lstatIfPresent(path) { + try { + return lstatSync(path); + } catch (error) { + if (error.code === "ENOENT") return undefined; + throw error; + } +} + +function readRegularFile(path, label) { + const status = lstatIfPresent(path); + if (!status || !status.isFile() || status.isSymbolicLink()) { + throw new Error(`${label} is missing or is not a regular file: ${path}`); + } + return readFileSync(path); +} + +function markerExists(transactionPath, name) { + const path = resolve(transactionPath, name); + const status = lstatIfPresent(path); + if (!status) return false; + if (!status.isFile() || status.isSymbolicLink()) { + throw new Error(`Invalid upstream sync transaction marker: ${path}`); + } + return true; +} + +function fingerprint(path) { + const status = lstatIfPresent(path); + if (!status) return { kind: "absent" }; + if (!status.isFile()) throw new Error(`Upstream sync target is not a regular file: ${path}`); + return { + kind: "file", + sha256: digest(readFileSync(path)), + mode: statSync(path).mode & 0o777, + }; +} + +export function captureUpstreamSyncFingerprint(path) { + return fingerprint(path); +} + +function fingerprintsEqual(left, right) { + return left.kind === right.kind && + (left.kind === "absent" || (left.sha256 === right.sha256 && left.mode === right.mode)); +} + +function describeFingerprint(value) { + return value.kind === "absent" ? "absent" : `${value.sha256} mode ${value.mode.toString(8)}`; +} + +function assertFingerprint(path, expected, label) { + const actual = fingerprint(path); + if (!fingerprintsEqual(actual, expected)) { + throw new Error( + `${label} changed unexpectedly: ${path}\n` + + `Expected ${describeFingerprint(expected)}, found ${describeFingerprint(actual)}.`, + ); + } +} + +function normalizedRelativePath(value, label) { + if (typeof value !== "string" || value.length === 0 || value.includes("\\")) { + throw new Error(`Invalid upstream sync transaction ${label}.`); + } + if (posix.isAbsolute(value) || isAbsolute(value)) { + throw new Error(`Upstream sync transaction ${label} must stay inside the target checkout: ${value}`); + } + const normalized = posix.normalize(value); + if (normalized !== value || value === "." || value === ".." || value.startsWith("../")) { + throw new Error(`Upstream sync transaction ${label} must stay inside the target checkout: ${value}`); + } + return value; +} + +function resolveInside(root, path, label) { + const normalized = normalizedRelativePath(path, label); + const candidate = resolve(root, ...normalized.split("/")); + const fromRoot = relative(root, candidate); + if (fromRoot === ".." || fromRoot.startsWith(`..${sep}`) || isAbsolute(fromRoot)) { + throw new Error(`Upstream sync transaction ${label} resolves outside the target checkout: ${path}`); + } + + let existing = candidate; + while (!lstatIfPresent(existing)) { + const parent = dirname(existing); + if (parent === existing) { + throw new Error(`Cannot validate upstream sync transaction ${label}: ${path}`); + } + existing = parent; + } + const fromRealRoot = relative(realpathSync(root), realpathSync(existing)); + if (fromRealRoot === ".." || fromRealRoot.startsWith(`..${sep}`) || isAbsolute(fromRealRoot)) { + throw new Error(`Upstream sync transaction ${label} escapes the target checkout through a link: ${path}`); + } + return candidate; +} + +function sidecarPath(target, id, index, suffix) { + return posix.join(posix.dirname(target), SIDECAR_NAME, id, `${index}.${suffix}`); +} + +function localPathKey(path) { + return process.platform === "win32" || process.platform === "darwin" ? path.toLowerCase() : path; +} + +function pathsOverlap(left, right) { + const leftKey = localPathKey(left); + const rightKey = localPathKey(right); + return leftKey === rightKey || leftKey.startsWith(`${rightKey}/`) || rightKey.startsWith(`${leftKey}/`); +} + +function assertTargetDoesNotOverlap(targets, target, label) { + const overlap = targets.find((candidate) => pathsOverlap(candidate, target)); + if (overlap) throw new Error(`Overlapping targets in ${label}: ${overlap} and ${target}`); +} + +function assertFingerprintShape(value, label) { + if (!value || typeof value !== "object" || !["absent", "file"].includes(value.kind)) { + throw new Error(`Invalid upstream sync transaction ${label}.`); + } + if (value.kind === "file" && + (!HASH_PATTERN.test(value.sha256) || !Number.isInteger(value.mode) || value.mode < 0 || value.mode > 0o777)) { + throw new Error(`Invalid upstream sync transaction ${label}.`); + } +} + +function transactionRoots(targetRoot) { + const stateRoot = resolve(targetRoot, STATE_NAME); + const transactionsRoot = resolve(stateRoot, "transactions"); + for (const [path, label] of [[stateRoot, "state root"], [transactionsRoot, "transactions root"]]) { + const status = lstatIfPresent(path); + if (status && (!status.isDirectory() || status.isSymbolicLink())) { + throw new Error(`Invalid upstream sync ${label}: ${path}`); + } + if (status) { + const fromTarget = relative(realpathSync(targetRoot), realpathSync(path)); + if (fromTarget === ".." || fromTarget.startsWith(`..${sep}`) || isAbsolute(fromTarget)) { + throw new Error(`Upstream sync ${label} escapes the target checkout: ${path}`); + } + } + } + return { + stateRoot, + transactionsRoot, + }; +} + +function readJournal(targetRoot, transactionPath, id) { + const journalPath = resolve(transactionPath, "journal.json"); + let journal; + try { + journal = JSON.parse(readRegularFile(journalPath, "Upstream sync transaction journal").toString("utf8")); + } catch (error) { + throw new Error(`Invalid upstream sync transaction journal at ${journalPath}: ${error.message}`, { cause: error }); + } + + if (!journal || typeof journal !== "object" || journal.version !== JOURNAL_VERSION || journal.id !== id) { + throw new Error(`Invalid upstream sync transaction journal at ${journalPath}.`); + } + if (!journal.owner || !Number.isSafeInteger(journal.owner.pid) || journal.owner.pid < 1 || + typeof journal.owner.token !== "string" || journal.owner.token.length === 0) { + throw new Error(`Invalid upstream sync transaction owner at ${journalPath}.`); + } + if (journal.targetRoot !== realpathSync(targetRoot)) { + throw new Error(`Upstream sync transaction journal targets a different checkout: ${journalPath}`); + } + if (!Array.isArray(journal.createdDirectories) || !Array.isArray(journal.operations) || !journal.operations.length) { + throw new Error(`Invalid upstream sync transaction journal at ${journalPath}.`); + } + + const createdDirectories = new Set(); + for (const [index, path] of journal.createdDirectories.entries()) { + normalizedRelativePath(path, `createdDirectories[${index}]`); + resolveInside(targetRoot, path, `createdDirectories[${index}]`); + if (createdDirectories.has(path)) throw new Error(`Duplicate directory in upstream sync transaction: ${path}`); + createdDirectories.add(path); + } + + const targets = []; + for (const [index, operation] of journal.operations.entries()) { + if (!operation || typeof operation !== "object" || + !["write", "remove", "manifest"].includes(operation.kind)) { + throw new Error(`Invalid upstream sync transaction operation ${index}.`); + } + normalizedRelativePath(operation.target, `operations[${index}].target`); + resolveInside(targetRoot, operation.target, `operations[${index}].target`); + assertTargetDoesNotOverlap(targets, operation.target, "upstream sync transaction"); + targets.push(operation.target); + + const expectedBackup = sidecarPath(operation.target, id, index, "old"); + const expectedStage = operation.kind === "remove" ? null : sidecarPath(operation.target, id, index, "new"); + if (operation.backup !== expectedBackup || operation.stage !== expectedStage) { + throw new Error(`Invalid sidecar path in upstream sync transaction operation ${index}.`); + } + resolveInside(targetRoot, operation.backup, `operations[${index}].backup`); + if (operation.stage !== null) resolveInside(targetRoot, operation.stage, `operations[${index}].stage`); + assertFingerprintShape(operation.before, `operations[${index}].before`); + assertFingerprintShape(operation.after, `operations[${index}].after`); + if (operation.kind === "remove" && operation.after.kind !== "absent") { + throw new Error(`Invalid removal state in upstream sync transaction operation ${index}.`); + } + if (operation.kind !== "remove" && operation.after.kind !== "file") { + throw new Error(`Invalid write state in upstream sync transaction operation ${index}.`); + } + } + + const manifest = journal.operations.at(-1); + if (manifest.kind !== "manifest" || manifest.target !== "profiles/upstream-manifest.json") { + throw new Error(`Upstream sync transaction journal does not commit the manifest last: ${journalPath}`); + } + return journal; +} + +function listTransactions(targetRoot) { + const { transactionsRoot } = transactionRoots(targetRoot); + if (!existsSync(transactionsRoot)) return []; + return readdirSync(transactionsRoot, { withFileTypes: true }) + .map((entry) => { + if (!entry.isDirectory()) { + throw new Error(`Invalid entry in upstream sync transaction directory: ${resolve(transactionsRoot, entry.name)}`); + } + return entry.name; + }) + .sort(); +} + +function removeEmptyDirectory(path) { + try { + rmdirSync(path); + fsyncDirectory(dirname(path)); + } catch (error) { + if (!["ENOENT", "ENOTEMPTY", "EEXIST"].includes(error.code)) throw error; + } +} + +function removeIfPresent(path) { + const status = lstatIfPresent(path); + if (!status) return; + if (!status.isFile() || status.isSymbolicLink()) { + throw new Error(`Refusing to remove non-file upstream sync state: ${path}`); + } + durableUnlink(path); +} + +function removeExpectedFile(path, expected, label) { + if (!lstatIfPresent(path)) return; + assertFingerprint(path, expected, label); + durableUnlink(path); +} + +function removeOwnedEmptyDirectory(path) { + const status = lstatIfPresent(path); + if (!status) return; + if (!status.isDirectory() || status.isSymbolicLink()) { + throw new Error(`Invalid upstream sync sidecar directory: ${path}`); + } + const entries = readdirSync(path); + if (entries.length) { + throw new Error(`Unexpected content in upstream sync sidecar directory: ${path}\n${entries.join("\n")}`); + } + rmdirSync(path); + fsyncDirectory(dirname(path)); +} + +function cleanupEmptyStateRoots(targetRoot) { + const { stateRoot, transactionsRoot } = transactionRoots(targetRoot); + removeEmptyDirectory(transactionsRoot); + removeEmptyDirectory(stateRoot); +} + +function cleanupAbandonedJournalWrite(targetRoot, transactionPath) { + const entries = readdirSync(transactionPath); + const completedCleanup = entries.includes("COMMITTED") && + entries.every((entry) => entry === "COMMITTED" || entry === "COMMITTED.tmp"); + const abandonedJournal = entries.every((entry) => entry === "journal.json.tmp"); + if (!completedCleanup && !abandonedJournal) { + throw new Error(`Invalid incomplete upstream sync transaction at ${transactionPath}.`); + } + removeIfPresent(resolve(transactionPath, "COMMITTED.tmp")); + removeIfPresent(resolve(transactionPath, "COMMITTED")); + removeIfPresent(resolve(transactionPath, "journal.json.tmp")); + removeEmptyDirectory(transactionPath); + cleanupEmptyStateRoots(targetRoot); +} + +function transactionPaths(targetRoot, journal) { + return journal.operations.map((operation) => ({ + ...operation, + targetPath: resolveInside(targetRoot, operation.target, "operation target"), + stagePath: operation.stage === null ? undefined : resolveInside(targetRoot, operation.stage, "operation stage"), + backupPath: resolveInside(targetRoot, operation.backup, "operation backup"), + })); +} + +function rollbackOperation(operation) { + const backup = lstatIfPresent(operation.backupPath); + const current = fingerprint(operation.targetPath); + + if (operation.before.kind === "file") { + if (backup) { + assertFingerprint(operation.backupPath, operation.before, "Upstream sync backup"); + if (current.kind === "absent") { + durableRename(operation.backupPath, operation.targetPath); + } else if (fingerprintsEqual(current, operation.after)) { + durableUnlink(operation.targetPath); + durableRename(operation.backupPath, operation.targetPath); + } else if (fingerprintsEqual(current, operation.before)) { + durableUnlink(operation.backupPath); + } else { + throw new Error(`Refusing to overwrite an unknown file while rolling back: ${operation.targetPath}`); + } + } else if (!fingerprintsEqual(current, operation.before)) { + throw new Error(`Cannot restore upstream sync target because its backup is missing: ${operation.targetPath}`); + } + } else { + if (backup) throw new Error(`Unexpected backup for a newly created target: ${operation.backupPath}`); + if (fingerprintsEqual(current, operation.after)) durableUnlink(operation.targetPath); + else if (current.kind !== "absent") { + throw new Error(`Refusing to remove an unknown file while rolling back: ${operation.targetPath}`); + } + } + + if (operation.stagePath) removeExpectedFile(operation.stagePath, operation.after, "Upstream sync staged file"); + assertFingerprint(operation.targetPath, operation.before, "Rolled back upstream sync target"); +} + +function rollbackJournal(targetRoot, journal) { + const errors = []; + for (const operation of transactionPaths(targetRoot, journal).reverse()) { + try { + rollbackOperation(operation); + } catch (error) { + errors.push(`${operation.target}: ${error.message}`); + } + } + if (errors.length) throw new Error(errors.join("\n")); +} + +function cleanupTransaction(targetRoot, transactionPath, journal) { + const sidecarTransactions = new Set(); + for (const operation of transactionPaths(targetRoot, journal)) { + if (operation.stagePath) removeExpectedFile(operation.stagePath, operation.after, "Upstream sync staged file"); + if (lstatIfPresent(operation.backupPath)) { + assertFingerprint(operation.backupPath, operation.before, "Upstream sync backup"); + durableUnlink(operation.backupPath); + } + sidecarTransactions.add(dirname(operation.backupPath)); + } + + for (const path of sidecarTransactions) { + removeOwnedEmptyDirectory(path); + removeEmptyDirectory(dirname(path)); + } + for (const path of [...journal.createdDirectories].reverse()) { + removeEmptyDirectory(resolveInside(targetRoot, path, "created directory")); + } + removeIfPresent(resolve(transactionPath, "COMMITTING.tmp")); + removeIfPresent(resolve(transactionPath, "COMMITTING")); + removeIfPresent(resolve(transactionPath, "COMMITTED.tmp")); + const allowedEntries = new Set(["COMMITTED", "journal.json", "journal.json.tmp"]); + const unknownEntries = readdirSync(transactionPath).filter((entry) => !allowedEntries.has(entry)); + if (unknownEntries.length) { + throw new Error(`Unexpected content in upstream sync transaction directory: ${transactionPath}\n${unknownEntries.join("\n")}`); + } + removeIfPresent(resolve(transactionPath, "journal.json")); + removeIfPresent(resolve(transactionPath, "journal.json.tmp")); + // Keep COMMITTED until the journal is gone so cleanup cannot be mistaken for rollback. + removeIfPresent(resolve(transactionPath, "COMMITTED")); + removeEmptyDirectory(transactionPath); + cleanupEmptyStateRoots(targetRoot); +} + +function validateCommittedTargets(targetRoot, journal) { + for (const operation of transactionPaths(targetRoot, journal)) { + assertFingerprint(operation.targetPath, operation.after, "Committed upstream sync target"); + } +} + +function processIsAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return error.code !== "ESRCH"; + } +} + +function lockOwnerText(owner) { + return JSON.stringify(owner); +} + +function assertLockOwned(lock) { + let current; + try { + current = JSON.parse(readRegularFile(lock.lockPath, "Upstream sync lock").toString("utf8")); + } catch (error) { + throw new Error(`Upstream sync lost its lock: ${lock.lockPath}`, { cause: error }); + } + if (current.pid !== lock.owner.pid || current.token !== lock.owner.token) { + throw new Error(`Upstream sync lost its lock to another process: ${lock.lockPath}`); + } +} + +function createOwnedLock(lockPath, owner) { + const temporaryPath = `${lockPath}.${owner.pid}.${owner.token}.tmp`; + let published = false; + try { + writeDurableFile(temporaryPath, `${lockOwnerText(owner)}\n`); + linkSync(temporaryPath, lockPath); + published = true; + fsyncDirectory(dirname(lockPath)); + durableUnlink(temporaryPath); + return owner; + } catch (error) { + try { + if (published && + readRegularFile(lockPath, "Upstream sync lock").toString("utf8").trim() === lockOwnerText(owner)) { + durableUnlink(lockPath); + } + } catch { + // Keep the original lock creation error. + } + try { + if (existsSync(temporaryPath)) durableUnlink(temporaryPath); + } catch { + // Keep the original lock creation error. + } + throw error; + } +} + +function removeDeadLockInitializers(targetRoot) { + const prefix = `${LOCK_NAME}.`; + for (const entry of readdirSync(targetRoot, { withFileTypes: true })) { + if (!entry.isFile() || !entry.name.startsWith(prefix) || !entry.name.endsWith(".tmp")) continue; + const match = entry.name.match(/^\.mstack-sync-upstream\.lock\.([1-9]\d*)\.([0-9a-f-]+)\.tmp$/); + if (!match) throw new Error(`Invalid upstream sync lock initializer: ${resolve(targetRoot, entry.name)}`); + const pid = Number(match[1]); + if (!Number.isSafeInteger(pid) || processIsAlive(pid)) { + throw new Error(`Another upstream sync is initializing its lock: ${resolve(targetRoot, entry.name)}`); + } + try { + durableUnlink(resolve(targetRoot, entry.name)); + } catch (error) { + if (error.code !== "ENOENT") throw error; + } + } +} + +function acquireTakeoverGuard(guardPath, owner) { + const reclaimPrefix = `${basename(guardPath)}.`; + for (const entry of readdirSync(dirname(guardPath), { withFileTypes: true })) { + if (!entry.isFile() || !entry.name.startsWith(reclaimPrefix) || !entry.name.endsWith(".reclaimed")) continue; + const reclaimedPath = resolve(dirname(guardPath), entry.name); + let reclaimedOwner; + try { + reclaimedOwner = JSON.parse(readRegularFile(reclaimedPath, "upstream sync takeover claim").toString("utf8")); + } catch (error) { + throw new Error(`Invalid upstream sync takeover claim: ${reclaimedPath}`, { cause: error }); + } + if (!reclaimedOwner || !Number.isSafeInteger(reclaimedOwner.pid) || reclaimedOwner.pid < 1 || + processIsAlive(reclaimedOwner.pid)) { + throw new Error(`Another upstream sync is recovering the lock: ${reclaimedPath}`); + } + durableUnlink(reclaimedPath); + } + try { + return { lockPath: guardPath, owner: createOwnedLock(guardPath, owner) }; + } catch (error) { + if (error.code !== "EEXIST") throw error; + } + let current; + try { + current = JSON.parse(readRegularFile(guardPath, "upstream sync takeover guard").toString("utf8")); + } catch (error) { + throw new Error(`Invalid or initializing upstream sync takeover guard: ${guardPath}`, { cause: error }); + } + if (!current || current.version !== 1 || !Number.isSafeInteger(current.pid) || current.pid < 1 || + typeof current.token !== "string" || current.token.length === 0 || processIsAlive(current.pid)) { + throw new Error(`Another upstream sync is recovering the lock: ${guardPath}`); + } + const reclaimedPath = `${guardPath}.${owner.token}.reclaimed`; + try { + durableRename(guardPath, reclaimedPath); + } catch (error) { + if (error.code === "ENOENT" || error.code === "EEXIST") { + throw new Error(`Another upstream sync is recovering the lock: ${guardPath}`, { cause: error }); + } + throw error; + } + try { + return { lockPath: guardPath, owner: createOwnedLock(guardPath, owner) }; + } catch (error) { + if (error.code === "EEXIST") throw new Error(`Another upstream sync is recovering the lock: ${guardPath}`, { cause: error }); + throw error; + } finally { + if (lstatIfPresent(reclaimedPath)) durableUnlink(reclaimedPath); + } +} + +export function acquireUpstreamSyncLock(targetRoot, { recoverStale = true } = {}) { + const lockPath = resolve(targetRoot, LOCK_NAME); + const owner = { version: 1, pid: process.pid, token: randomUUID() }; + removeDeadLockInitializers(targetRoot); + try { + return { lockPath, owner: createOwnedLock(lockPath, owner) }; + } catch (error) { + if (error.code !== "EEXIST") throw error; + } + + let current; + try { + current = JSON.parse(readRegularFile(lockPath, "Upstream sync lock").toString("utf8")); + } catch (error) { + throw new Error(`Invalid or initializing upstream sync lock: ${lockPath}`, { cause: error }); + } + if (!current || current.version !== 1 || !Number.isSafeInteger(current.pid) || current.pid < 1 || + typeof current.token !== "string" || current.token.length === 0) { + throw new Error(`Invalid upstream sync lock: ${lockPath}`); + } + if (processIsAlive(current.pid)) { + throw new Error(`Another live upstream sync owns the lock: ${lockPath}`); + } + if (!recoverStale) { + throw new Error(`Target has an active or interrupted upstream sync; rerun sync-upstream with --apply: ${lockPath}`); + } + + const guardPath = `${lockPath}.takeover`; + let guard; + try { + guard = acquireTakeoverGuard(guardPath, owner); + let guardedCurrent; + try { + guardedCurrent = JSON.parse(readRegularFile(lockPath, "Upstream sync lock").toString("utf8")); + } catch (error) { + throw new Error(`Invalid or initializing upstream sync lock: ${lockPath}`, { cause: error }); + } + if (!guardedCurrent || guardedCurrent.version !== 1 || !Number.isSafeInteger(guardedCurrent.pid) || + guardedCurrent.pid < 1 || typeof guardedCurrent.token !== "string" || guardedCurrent.token.length === 0) { + throw new Error(`Invalid upstream sync lock: ${lockPath}`); + } + if (processIsAlive(guardedCurrent.pid)) { + throw new Error(`Another live upstream sync owns the lock: ${lockPath}`); + } + durableUnlink(lockPath); + try { + return { lockPath, owner: createOwnedLock(lockPath, owner) }; + } catch (error) { + if (error.code === "EEXIST") throw new Error(`Another upstream sync became active: ${lockPath}`, { cause: error }); + throw error; + } + } finally { + if (guard) releaseUpstreamSyncLock(guard); + } +} + +export function releaseUpstreamSyncLock(lock) { + let current; + try { + current = JSON.parse(readRegularFile(lock.lockPath, "Upstream sync lock").toString("utf8")); + } catch (error) { + if (error.code === "ENOENT") return; + throw error; + } + if (current.pid !== lock.owner.pid || current.token !== lock.owner.token) { + throw new Error(`Refusing to release an upstream sync lock owned by another process: ${lock.lockPath}`); + } + durableUnlink(lock.lockPath); +} + +export function assertUpstreamSyncTargetReadable(targetRoot, heldLock) { + if (heldLock) assertLockOwned(heldLock); + const lockPath = resolve(targetRoot, LOCK_NAME); + const transactions = listTransactions(targetRoot); + if ((!heldLock && existsSync(lockPath)) || transactions.length) { + throw new Error(`Target has an active or interrupted upstream sync; rerun sync-upstream with --apply: ${targetRoot}`); + } +} + +export function recoverUpstreamSyncTransactions(targetRoot, lock) { + assertLockOwned(lock); + let recovered = 0; + for (const id of listTransactions(targetRoot)) { + assertLockOwned(lock); + normalizedRelativePath(id, "transaction id"); + if (id.includes("/")) throw new Error(`Invalid upstream sync transaction id: ${id}`); + const { transactionsRoot } = transactionRoots(targetRoot); + const transactionPath = resolve(transactionsRoot, id); + const journalPath = resolve(transactionPath, "journal.json"); + if (!existsSync(journalPath)) { + cleanupAbandonedJournalWrite(targetRoot, transactionPath); + recovered += 1; + continue; + } + + const journal = readJournal(targetRoot, transactionPath, id); + if (journal.owner.pid === process.pid && journal.owner.token !== lock.owner.token) { + throw new Error(`Upstream sync transaction owner token does not match the current lock: ${transactionPath}`); + } + if (journal.owner.pid !== process.pid && processIsAlive(journal.owner.pid)) { + throw new Error(`Upstream sync transaction is still owned by live process ${journal.owner.pid}: ${transactionPath}`); + } + if (markerExists(transactionPath, "COMMITTED")) validateCommittedTargets(targetRoot, journal); + else rollbackJournal(targetRoot, journal); + cleanupTransaction(targetRoot, transactionPath, journal); + recovered += 1; + } + assertLockOwned(lock); + return recovered; +} + +function missingParentDirectories(targetRoot, targetPath) { + const missing = []; + let current = dirname(targetPath); + while (current !== targetRoot) { + const status = lstatIfPresent(current); + if (status) { + if (!status.isDirectory()) throw new Error(`Upstream sync target parent is not a directory: ${current}`); + break; + } + missing.push(current); + current = dirname(current); + } + return missing.reverse(); +} + +function buildJournal(targetRoot, id, changes, lock) { + if (!Array.isArray(changes) || !changes.length) throw new Error("Upstream sync transaction has no operations."); + const targets = []; + const createdDirectories = new Set(); + const operations = changes.map((change, index) => { + if (!change || !["write", "remove", "manifest"].includes(change.kind)) { + throw new Error(`Invalid upstream sync change ${index}.`); + } + const target = normalizedRelativePath(change.target, `change ${index} target`); + if (target.split("/").some((part) => [LOCK_NAME, STATE_NAME, SIDECAR_NAME].includes(part))) { + throw new Error(`Upstream sync target uses a reserved transaction path: ${target}`); + } + assertTargetDoesNotOverlap(targets, target, "upstream sync transaction"); + targets.push(target); + const targetPath = resolveInside(targetRoot, target, `change ${index} target`); + for (const path of missingParentDirectories(targetRoot, targetPath)) { + createdDirectories.add(relative(targetRoot, path).split(sep).join("/")); + } + + const before = fingerprint(targetPath); + assertFingerprintShape(change.before, `change ${index} before`); + if (!fingerprintsEqual(before, change.before)) { + throw new Error(`Upstream sync target changed after planning: ${targetPath}`); + } + let after; + if (change.kind === "remove") { + if (change.content !== undefined) throw new Error(`Removal has unexpected content: ${target}`); + after = { kind: "absent" }; + } else { + if (!Buffer.isBuffer(change.content)) throw new Error(`Write has no Buffer content: ${target}`); + after = { + kind: "file", + sha256: digest(change.content), + mode: before.kind === "file" ? before.mode : 0o666 & ~process.umask(), + }; + } + return { + kind: change.kind, + target, + stage: change.kind === "remove" ? null : sidecarPath(target, id, index, "new"), + backup: sidecarPath(target, id, index, "old"), + before, + after, + }; + }); + + const manifest = operations.at(-1); + if (manifest.kind !== "manifest" || manifest.target !== "profiles/upstream-manifest.json") { + throw new Error("Upstream sync transaction must write the manifest last."); + } + return { + version: JOURNAL_VERSION, + id, + owner: lock.owner, + targetRoot: realpathSync(targetRoot), + createdDirectories: [...createdDirectories], + operations, + }; +} + +function createJournal(targetRoot, journal) { + const { stateRoot, transactionsRoot } = transactionRoots(targetRoot); + const transactionPath = resolve(transactionsRoot, journal.id); + mkdirSync(transactionPath, { recursive: true, mode: 0o700 }); + transactionRoots(targetRoot); + const transactionStatus = lstatSync(transactionPath); + const fromTarget = relative(realpathSync(targetRoot), realpathSync(transactionPath)); + if (!transactionStatus.isDirectory() || transactionStatus.isSymbolicLink() || + fromTarget === ".." || fromTarget.startsWith(`..${sep}`) || isAbsolute(fromTarget)) { + throw new Error(`Invalid upstream sync transaction directory: ${transactionPath}`); + } + fsyncDirectory(targetRoot); + fsyncDirectory(stateRoot); + fsyncDirectory(transactionsRoot); + const temporaryPath = resolve(transactionPath, "journal.json.tmp"); + const journalPath = resolve(transactionPath, "journal.json"); + writeDurableFile(temporaryPath, `${JSON.stringify(journal, null, 2)}\n`); + durableRename(temporaryPath, journalPath); + return transactionPath; +} + +function createTargetDirectories(targetRoot, journal) { + for (const path of journal.createdDirectories) { + const directory = resolveInside(targetRoot, path, "created directory"); + const status = lstatIfPresent(directory); + if (status) { + throw new Error(`Upstream sync target directory changed after planning: ${directory}`); + } + mkdirSync(directory); + fsyncDirectory(dirname(directory)); + } +} + +function prepareStages(targetRoot, journal, changes, lock) { + const operations = transactionPaths(targetRoot, journal); + for (const [index, operation] of operations.entries()) { + assertLockOwned(lock); + const sidecarTransaction = dirname(operation.backupPath); + const sidecarRoot = dirname(sidecarTransaction); + for (const path of [sidecarRoot, sidecarTransaction]) { + const status = lstatIfPresent(path); + if (status && (!status.isDirectory() || status.isSymbolicLink())) { + throw new Error(`Invalid upstream sync sidecar directory: ${path}`); + } + } + mkdirSync(sidecarTransaction, { recursive: true, mode: 0o700 }); + for (const path of [sidecarRoot, sidecarTransaction]) { + const status = lstatIfPresent(path); + if (!status?.isDirectory() || status.isSymbolicLink()) { + throw new Error(`Invalid upstream sync sidecar directory: ${path}`); + } + } + resolveInside(targetRoot, operation.backup, "operation backup"); + fsyncDirectory(dirname(sidecarRoot)); + fsyncDirectory(sidecarRoot); + if (!operation.stagePath) continue; + writeDurableFile(operation.stagePath, changes[index].content, operation.after.mode); + assertFingerprint(operation.stagePath, operation.after, "Staged upstream sync file"); + } +} + +function commitJournal(targetRoot, transactionPath, journal, lock) { + const operations = transactionPaths(targetRoot, journal); + for (const operation of operations) { + assertFingerprint(operation.targetPath, operation.before, "Upstream sync target before commit"); + if (lstatIfPresent(operation.backupPath)) { + throw new Error(`Upstream sync backup already exists: ${operation.backupPath}`); + } + } + + assertLockOwned(lock); + createDurableMarker(transactionPath, "COMMITTING", "committing\n"); + for (const operation of operations) { + assertLockOwned(lock); + if (operation.before.kind === "file") durableRename(operation.targetPath, operation.backupPath); + if (operation.stagePath) durableRename(operation.stagePath, operation.targetPath); + assertFingerprint(operation.targetPath, operation.after, "Upstream sync target after commit"); + } + validateCommittedTargets(targetRoot, journal); + assertLockOwned(lock); + createDurableMarker(transactionPath, "COMMITTED", "committed\n"); +} + +export function applyUpstreamSyncTransaction(targetRoot, changes, lock) { + assertLockOwned(lock); + const id = randomUUID(); + const journal = buildJournal(targetRoot, id, changes, lock); + const { transactionsRoot } = transactionRoots(targetRoot); + const transactionPath = resolve(transactionsRoot, id); + try { + createJournal(targetRoot, journal); + assertLockOwned(lock); + createTargetDirectories(targetRoot, journal); + prepareStages(targetRoot, journal, changes, lock); + commitJournal(targetRoot, transactionPath, journal, lock); + cleanupTransaction(targetRoot, transactionPath, journal); + } catch (error) { + if (markerExists(transactionPath, "COMMITTED")) { + throw new Error(`Upstream sync committed, but transaction cleanup failed: ${error.message}`, { cause: error }); + } + try { + if (existsSync(resolve(transactionPath, "journal.json"))) { + rollbackJournal(targetRoot, journal); + cleanupTransaction(targetRoot, transactionPath, journal); + } else if (existsSync(transactionPath)) { + cleanupAbandonedJournalWrite(targetRoot, transactionPath); + } + } catch (rollbackError) { + throw new Error( + `${error.message}\nRollback failed; transaction state was retained for recovery:\n${rollbackError.message}`, + { cause: error }, + ); + } + throw error; + } +} diff --git a/scripts/sync-upstream.mjs b/scripts/sync-upstream.mjs index cd76025..b74122f 100644 --- a/scripts/sync-upstream.mjs +++ b/scripts/sync-upstream.mjs @@ -3,17 +3,22 @@ import { existsSync, lstatSync, - mkdirSync, readFileSync, realpathSync, readdirSync, - unlinkSync, - writeFileSync, } from "node:fs"; import { execFileSync } from "node:child_process"; import { createHash } from "node:crypto"; import { dirname, extname, isAbsolute, join, posix, relative, resolve, sep, win32 } from "node:path"; import { fileURLToPath } from "node:url"; +import { + acquireUpstreamSyncLock, + applyUpstreamSyncTransaction, + assertUpstreamSyncTargetReadable, + captureUpstreamSyncFingerprint, + recoverUpstreamSyncTransactions, + releaseUpstreamSyncLock, +} from "./sync-upstream-transaction.mjs"; const args = process.argv.slice(2); const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); @@ -41,17 +46,33 @@ if (!source) { } const sourceRoot = resolve(source); -if (!existsSync(sourceRoot)) throw new Error(`Upstream checkout not found: ${sourceRoot}`); const targetRoot = resolve(valueAfter("--target") ?? repoRoot); -const profilesRoot = resolveInside(targetRoot, "profiles", "profiles directory"); +const apply = args.includes("--apply"); +const force = args.includes("--force"); +if (force && !apply) throw new Error("--force requires --apply"); + +const syncLock = acquireUpstreamSyncLock(targetRoot, { recoverStale: apply }); +const releaseReadLockAtExit = () => { + try { + releaseUpstreamSyncLock(syncLock); + } catch { + // A stale read lock is recoverable by the next apply. + } +}; +if (!apply) process.once("exit", releaseReadLockAtExit); + +try { +const recoveredTransactions = apply ? recoverUpstreamSyncTransactions(targetRoot, syncLock) : 0; +if (!apply) assertUpstreamSyncTargetReadable(targetRoot, syncLock); +if (recoveredTransactions) { + console.log(`Recovered ${recoveredTransactions} interrupted upstream sync transaction(s).`); +} +if (!existsSync(sourceRoot)) throw new Error(`Upstream checkout not found: ${sourceRoot}`); const upstreamsPath = resolveInside(targetRoot, "profiles/upstreams.json", "upstream profile"); if (!existsSync(upstreamsPath)) throw new Error(`mstack upstream profile not found: ${upstreamsPath}`); const upstreams = JSON.parse(readFileSync(upstreamsPath, "utf8")); const pstack = upstreams.pstack; if (!pstack || typeof pstack !== "object") throw new Error("profiles/upstreams.json has no pstack entry"); -const apply = args.includes("--apply"); -const force = args.includes("--force"); -if (force && !apply) throw new Error("--force requires --apply"); function gitOutput(arguments_) { try { @@ -288,15 +309,17 @@ for (const [name, configured] of Object.entries(artifacts)) { } expectedTargets.set(targetRelative, `${name}:${normalizePath(sourceFile)}`); const expected = transformedBuffer(sourcePath, spec.replacements); - const current = existsSync(targetPath) ? readFileSync(targetPath) : undefined; + const before = captureUpstreamSyncFingerprint(targetPath); + const current = before.kind === "file" ? readFileSync(targetPath) : undefined; let kind = "unchanged"; if (!current) kind = "new"; else if (!equalContent(targetPath, current, expected)) kind = "changed"; - operations.push({ name, spec, sourceFile, sourcePath, targetPath, expected, kind }); + operations.push({ name, spec, sourceFile, sourcePath, targetPath, expected, before, kind }); } } const manifestPath = resolveInside(targetRoot, "profiles/upstream-manifest.json", "upstream manifest"); +const manifestBefore = captureUpstreamSyncFingerprint(manifestPath); let previousManifest; if (existsSync(manifestPath)) { try { @@ -328,6 +351,7 @@ for (const [targetRelative, previous] of previousTargets) { if (expectedTargets.has(targetRelative) || missingSourceNames.has(previous.name)) continue; const targetPath = resolveInside(targetRoot, targetRelative, `upstream manifest artifact ${previous.name}`); if (!existsSync(targetPath)) continue; + const before = captureUpstreamSyncFingerprint(targetPath); const current = readFileSync(targetPath); const matchesBaseline = digest( textExtensions.has(extname(targetPath).toLowerCase()) @@ -339,6 +363,7 @@ for (const [targetRelative, previous] of previousTargets) { kind: "removed", targetPath, targetRelative, + before, matchesBaseline, }); } @@ -418,19 +443,12 @@ if ((conflicts.length || removalConflicts.length) && !force) { ); } -const written = []; -for (const operation of operations) { - if (operation.kind === "unchanged" || operation.kind === "removed") continue; - if (operation.kind === "changed" && operation.spec?.compareContent === false && !force) continue; - mkdirSync(dirname(operation.targetPath), { recursive: true }); - writeFileSync(operation.targetPath, operation.expected); - written.push(operation); -} -const removed = []; -for (const operation of removals) { - unlinkSync(operation.targetPath); - removed.push(operation); -} +const written = operations.filter((operation) => + operation.kind !== "unchanged" && + operation.kind !== "removed" && + (operation.kind !== "changed" || operation.spec?.compareContent !== false || force) +); +const removed = removals; const manifest = { source: pstack.repository, @@ -453,7 +471,29 @@ const manifest = { ]), ), }; -writeFileSync(manifestPath, `${JSON.stringify(manifest, null, 2)}\n`, "utf8"); +applyUpstreamSyncTransaction(targetRoot, [ + ...written.map((operation) => ({ + kind: "write", + target: normalizePath(relative(targetRoot, operation.targetPath)), + content: operation.expected, + before: operation.before, + })), + ...removed.map((operation) => ({ + kind: "remove", + target: operation.targetRelative, + before: operation.before, + })), + { + kind: "manifest", + target: "profiles/upstream-manifest.json", + content: Buffer.from(`${JSON.stringify(manifest, null, 2)}\n`, "utf8"), + before: manifestBefore, + }, +], syncLock); console.log(`Applied ${written.length} file(s).`); console.log(`Removed ${removed.length} file(s).`); console.log(`Wrote ${relative(targetRoot, manifestPath)}.`); +} finally { + releaseUpstreamSyncLock(syncLock); + if (!apply) process.removeListener("exit", releaseReadLockAtExit); +} diff --git a/scripts/sync-upstream.test.mjs b/scripts/sync-upstream.test.mjs index 8d1b8e1..dac6935 100644 --- a/scripts/sync-upstream.test.mjs +++ b/scripts/sync-upstream.test.mjs @@ -1,16 +1,19 @@ import assert from "node:assert/strict"; import { existsSync, + lstatSync, mkdirSync, mkdtempSync, readFileSync, + readdirSync, rmSync, unlinkSync, writeFileSync, } from "node:fs"; -import { spawnSync } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; +import { createHash } from "node:crypto"; import { tmpdir } from "node:os"; -import { join, resolve } from "node:path"; +import { join, relative, resolve } from "node:path"; import test from "node:test"; const synchronizer = resolve("scripts", "sync-upstream.mjs"); @@ -28,6 +31,205 @@ function run(script, arguments_) { }); } +const faultPreload = String.raw` +const fs = require("node:fs"); +const { syncBuiltinESMExports } = require("node:module"); + +const plan = JSON.parse(fs.readFileSync(process.env.MSTACK_SYNC_TEST_FAULT_PLAN, "utf8")); +const originals = Object.fromEntries( + ["closeSync", "copyFileSync", "existsSync", "linkSync", "mkdirSync", "openSync", "renameSync", "rmSync", "unlinkSync", "writeFileSync"] + .map((name) => [name, fs[name].bind(fs)]), +); +const descriptorPaths = new Map(); +const hits = new Map(); + +function normalized(value) { + if (typeof value === "number") return descriptorPaths.get(value) ?? String(value); + return String(value).replaceAll("\\", "/"); +} + +function ruleMatches(rule, paths) { + const [path, destination] = paths.map(normalized); + return (!rule.pathEndsWith || path.endsWith(rule.pathEndsWith)) && + (!rule.pathIncludes || path.includes(rule.pathIncludes)) && + (!rule.fromEndsWith || path.endsWith(rule.fromEndsWith)) && + (!rule.fromIncludes || path.includes(rule.fromIncludes)) && + (!rule.toEndsWith || destination?.endsWith(rule.toEndsWith)) && + (!rule.toIncludes || destination?.includes(rule.toIncludes)); +} + +function trigger(method, phase, paths) { + for (const [index, rule] of plan.rules.entries()) { + if (rule.method !== method || rule.phase !== phase || !ruleMatches(rule, paths)) continue; + const count = (hits.get(index) ?? 0) + 1; + hits.set(index, count); + if (count !== (rule.occurrence ?? 1)) continue; + + if (rule.action === "exit") process.exit(rule.exitCode ?? 86); + if (rule.action === "gate") { + originals.writeFileSync(rule.ready, "ready\n", "utf8"); + const deadline = Date.now() + (rule.timeoutMs ?? 10000); + const sleeper = new Int32Array(new SharedArrayBuffer(4)); + while (!originals.existsSync(rule.release)) { + if (Date.now() >= deadline) process.exit(87); + Atomics.wait(sleeper, 0, 0, 20); + } + return; + } + + const error = new Error(rule.message ?? "injected filesystem failure"); + error.code = rule.code ?? "EIO"; + throw error; + } +} + +for (const method of ["copyFileSync", "linkSync", "mkdirSync", "renameSync", "rmSync", "unlinkSync"]) { + fs[method] = (...args) => { + trigger(method, "before", args); + const result = originals[method](...args); + trigger(method, "after", args); + return result; + }; +} + +fs.openSync = (...args) => { + trigger("openSync", "before", args); + const descriptor = originals.openSync(...args); + descriptorPaths.set(descriptor, normalized(args[0])); + trigger("openSync", "after", args); + return descriptor; +}; + +fs.writeFileSync = (...args) => { + trigger("writeFileSync", "before", args); + const result = originals.writeFileSync(...args); + trigger("writeFileSync", "after", args); + return result; +}; + +fs.closeSync = (descriptor) => { + try { + return originals.closeSync(descriptor); + } finally { + descriptorPaths.delete(descriptor); + } +}; + +syncBuiltinESMExports(); +`; + +function writeFaultPlan(root, rules) { + const preloadPath = join(root, "sync-upstream-fault-preload.cjs"); + const planPath = join(root, "sync-upstream-fault-plan.json"); + write(preloadPath, faultPreload); + write(planPath, JSON.stringify({ rules })); + return { + preloadPath, + env: { ...process.env, MSTACK_SYNC_TEST_FAULT_PLAN: planPath }, + }; +} + +function runWithFaults(script, arguments_, root, rules, options = {}) { + const fault = writeFaultPlan(root, rules); + return spawnSync(process.execPath, ["--require", fault.preloadPath, script, ...arguments_], { + cwd: resolve("."), + encoding: "utf8", + env: fault.env, + timeout: options.timeout ?? 10000, + }); +} + +function spawnWithFaults(script, arguments_, root, rules) { + const fault = writeFaultPlan(root, rules); + const child = spawn(process.execPath, ["--require", fault.preloadPath, script, ...arguments_], { + cwd: resolve("."), + env: fault.env, + stdio: ["ignore", "pipe", "pipe"], + }); + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + let stdout = ""; + let stderr = ""; + child.stdout.on("data", (chunk) => { stdout += chunk; }); + child.stderr.on("data", (chunk) => { stderr += chunk; }); + const completed = new Promise((resolveResult, reject) => { + child.once("error", reject); + child.once("close", (status, signal) => resolveResult({ status, signal, stdout, stderr })); + }); + return { child, completed }; +} + +async function waitForPath(path, timeout = 5000) { + const deadline = Date.now() + timeout; + while (!existsSync(path)) { + if (Date.now() >= deadline) throw new Error(`Timed out waiting for ${path}`); + await new Promise((resolveWait) => setTimeout(resolveWait, 20)); + } +} + +function snapshotTree(root, { excludeTransactionState = false } = {}) { + const snapshot = []; + function visit(directory) { + for (const entry of readdirSync(directory, { withFileTypes: true }).sort((left, right) => left.name.localeCompare(right.name))) { + const path = join(directory, entry.name); + const relativePath = relative(root, path).replaceAll("\\", "/"); + if ( + excludeTransactionState && + (relativePath.startsWith(".mstack-sync-upstream.lock") || + relativePath === ".mstack-sync-upstream" || + relativePath.startsWith(".mstack-sync-upstream/") || + relativePath === ".mstack-sync-upstream-tx" || + relativePath.includes("/.mstack-sync-upstream-tx")) + ) continue; + if (entry.isDirectory()) { + snapshot.push(["directory", relativePath]); + visit(path); + } else { + const status = lstatSync(path); + snapshot.push([ + status.isSymbolicLink() ? "symlink" : "file", + relativePath, + status.isSymbolicLink() ? undefined : readFileSync(path).toString("base64"), + ]); + } + } + } + visit(root); + return snapshot; +} + +function assertNoTransactionState(target) { + const leftovers = snapshotTree(target) + .map(([, path]) => path) + .filter((path) => + path === ".mstack-sync-upstream.lock" || + path === ".mstack-sync-upstream" || + path.startsWith(".mstack-sync-upstream/") || + path === ".mstack-sync-upstream-tx" || + path.includes("/.mstack-sync-upstream-tx"), + ); + assert.deepEqual(leftovers, [], `transaction state was not removed: ${leftovers.join(", ")}`); +} + +function applyBaseline(source, target) { + const result = run(synchronizer, ["--source", source, "--target", target, "--apply"]); + assert.equal(result.status, 0, result.stderr); +} + +function advanceSource(source, target, update, message) { + update(); + const commit = commitSourceChange(source, message); + updateProfile(target, (pstack) => { pstack.commit = commit; }); +} + +function transactionDirectory(target, id) { + return join(target, ".mstack-sync-upstream", "transactions", id); +} + +function sha256(content) { + return createHash("sha256").update(content).digest("hex"); +} + function runGit(directory, arguments_) { const result = spawnSync("git", ["-C", directory, ...arguments_], { encoding: "utf8", @@ -356,3 +558,501 @@ test("locally adapted removed artifacts require force before deletion", () => { rmSync(root, { recursive: true, force: true }); } }); + +test("rolls back every target when a file replacement fails during commit", () => { + const { root, source, target } = fixture(); + try { + applyBaseline(source, target); + advanceSource(source, target, () => { + write(join(source, "automations", "benny", "README.md"), "changed pstack automation\n"); + write( + join(source, "docs", "guide", "02-poteto-mode.md"), + "Changed /poteto-mode guide for pstack.\nCursor confirms.\n", + ); + }, "change two managed files"); + const before = snapshotTree(target); + + const failed = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply", "--force"], + root, + [{ + method: "renameSync", + phase: "before", + fromIncludes: "/.mstack-sync-upstream-tx/", + toEndsWith: "/docs/guide/02-meta-mode.md", + message: "injected target replacement failure", + }], + ); + + assert.notEqual(failed.status, 0); + assert.match(failed.stderr, /injected target replacement failure/); + assert.deepEqual(snapshotTree(target), before); + assertNoTransactionState(target); + + const retried = run(synchronizer, ["--source", source, "--target", target, "--apply", "--force"]); + assert.equal(retried.status, 0, retried.stderr); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("rolls back file writes and removals when manifest replacement fails", () => { + const { root, source, target } = fixture(); + try { + applyBaseline(source, target); + advanceSource(source, target, () => { + unlinkSync(join(source, "automations", "benny", "README.md")); + write( + join(source, "docs", "guide", "02-poteto-mode.md"), + "Changed /poteto-mode guide for pstack.\nCursor confirms.\n", + ); + }, "change and remove managed files"); + const before = snapshotTree(target); + + const failed = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply", "--force"], + root, + [{ + method: "renameSync", + phase: "before", + fromIncludes: "/.mstack-sync-upstream-tx/", + toEndsWith: "/profiles/upstream-manifest.json", + message: "injected manifest replacement failure", + }], + ); + + assert.notEqual(failed.status, 0); + assert.match(failed.stderr, /injected manifest replacement failure/); + assert.deepEqual(snapshotTree(target), before); + assertNoTransactionState(target); + + const retried = run(synchronizer, ["--source", source, "--target", target, "--apply", "--force"]); + assert.equal(retried.status, 0, retried.stderr); + assert.equal(existsSync(join(target, "automations", "benny", "README.md")), false); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("retains recoverable state when rollback fails after a primary commit failure", () => { + const { root, source, target } = fixture(); + try { + applyBaseline(source, target); + advanceSource(source, target, () => { + write(join(source, "automations", "benny", "README.md"), "recovery pstack automation\n"); + write( + join(source, "docs", "guide", "02-poteto-mode.md"), + "Recovery /poteto-mode guide for pstack.\nCursor confirms.\n", + ); + }, "prepare rollback failure recovery"); + + const failed = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply", "--force"], + root, + [ + { + method: "renameSync", + phase: "before", + fromIncludes: "/.mstack-sync-upstream-tx/", + toEndsWith: "/profiles/upstream-manifest.json", + message: "injected primary manifest failure", + }, + { + method: "renameSync", + phase: "before", + fromIncludes: "/.mstack-sync-upstream-tx/", + fromEndsWith: ".old", + toEndsWith: "/automations/benny/README.md", + message: "injected rollback restore failure", + }, + ], + ); + + assert.notEqual(failed.status, 0); + assert.match(failed.stderr, /injected primary manifest failure/); + assert.match(failed.stderr, /Rollback failed; transaction state was retained for recovery/); + assert.match(failed.stderr, /injected rollback restore failure/); + assert.equal(existsSync(join(target, ".mstack-sync-upstream.lock")), false); + + const transactionsRoot = join(target, ".mstack-sync-upstream", "transactions"); + const transactionIds = readdirSync(transactionsRoot); + assert.equal(transactionIds.length, 1); + const transaction = join(transactionsRoot, transactionIds[0]); + const journalPath = join(transaction, "journal.json"); + assert.equal(existsSync(journalPath), true); + const journal = JSON.parse(readFileSync(journalPath, "utf8")); + const automation = journal.operations.find( + (operation) => operation.target === "automations/benny/README.md", + ); + assert.ok(automation, "journal did not retain the failed rollback operation"); + const backupPath = resolve(target, ...automation.backup.split("/")); + assert.equal(existsSync(backupPath), true, "rollback removed the only recoverable backup"); + + const recovered = run(synchronizer, ["--source", source, "--target", target, "--apply", "--force"]); + assert.equal(recovered.status, 0, recovered.stderr); + assert.equal( + readFileSync(join(target, "automations", "benny", "README.md"), "utf8"), + "recovery mstack automation\n", + ); + assert.equal( + readFileSync(join(target, "docs", "guide", "02-meta-mode.md"), "utf8"), + "Recovery /meta-mode guide for mstack.\nHarness confirms.\n", + ); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("recovers a stale transaction after the process exits between a rename and journal progress", () => { + const { root, source, target } = fixture(); + try { + applyBaseline(source, target); + advanceSource(source, target, () => { + write(join(source, "automations", "benny", "README.md"), "changed pstack automation\n"); + write( + join(source, "docs", "guide", "02-poteto-mode.md"), + "Changed /poteto-mode guide for pstack.\nCursor confirms.\n", + ); + }, "prepare interrupted sync"); + + const interrupted = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply", "--force"], + root, + [{ + method: "renameSync", + phase: "after", + fromIncludes: "/.mstack-sync-upstream-tx/", + toEndsWith: "/automations/benny/README.md", + action: "exit", + exitCode: 86, + }], + ); + + assert.equal(interrupted.status, 86, interrupted.stderr); + assert.equal(existsSync(join(target, ".mstack-sync-upstream.lock")), true); + const transactionsRoot = join(target, ".mstack-sync-upstream", "transactions"); + const transactionIds = readdirSync(transactionsRoot); + assert.equal(transactionIds.length, 1); + assert.equal(existsSync(join(transactionsRoot, transactionIds[0], "journal.json")), true); + + const recovered = run(synchronizer, ["--source", source, "--target", target, "--apply", "--force"]); + assert.equal(recovered.status, 0, recovered.stderr); + assert.equal( + readFileSync(join(target, "automations", "benny", "README.md"), "utf8"), + "changed mstack automation\n", + ); + assert.equal( + readFileSync(join(target, "docs", "guide", "02-meta-mode.md"), "utf8"), + "Changed /meta-mode guide for mstack.\nHarness confirms.\n", + ); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("cleans up a committed transaction without rolling it back after the committed marker rename", () => { + const { root, source, target } = fixture(); + try { + applyBaseline(source, target); + advanceSource(source, target, () => { + write(join(source, "automations", "benny", "README.md"), "committed pstack automation\n"); + write( + join(source, "docs", "guide", "02-poteto-mode.md"), + "Committed /poteto-mode guide for pstack.\nCursor confirms.\n", + ); + }, "prepare committed sync cleanup"); + + const interrupted = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply", "--force"], + root, + [{ + method: "renameSync", + phase: "after", + fromEndsWith: "/COMMITTED.tmp", + toEndsWith: "/COMMITTED", + action: "exit", + exitCode: 86, + }], + ); + + assert.equal(interrupted.status, 86, interrupted.stderr); + const transactionsRoot = join(target, ".mstack-sync-upstream", "transactions"); + const transactionIds = readdirSync(transactionsRoot); + assert.equal(transactionIds.length, 1); + assert.equal(existsSync(join(transactionsRoot, transactionIds[0], "COMMITTED")), true); + assert.equal( + readFileSync(join(target, "automations", "benny", "README.md"), "utf8"), + "committed mstack automation\n", + ); + + const recovered = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply", "--force"], + root, + [{ + method: "renameSync", + phase: "before", + fromIncludes: "/.mstack-sync-upstream-tx/", + toEndsWith: "/automations/benny/README.md", + message: "committed transaction attempted rollback", + }], + ); + assert.equal(recovered.status, 0, recovered.stderr); + assert.doesNotMatch(recovered.stderr, /committed transaction attempted rollback/); + assert.equal( + readFileSync(join(target, "automations", "benny", "README.md"), "utf8"), + "committed mstack automation\n", + ); + assert.equal( + readFileSync(join(target, "docs", "guide", "02-meta-mode.md"), "utf8"), + "Committed /meta-mode guide for mstack.\nHarness confirms.\n", + ); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("removes newly created nested target directories and sidecars after a first apply fails", () => { + const { root, source, target } = fixture(); + try { + updateProfile(target, (pstack) => { + pstack.artifacts.guide.target = "generated/deep/guide"; + }); + const before = snapshotTree(target); + + const failed = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply"], + root, + [{ + method: "renameSync", + phase: "before", + fromIncludes: "/.mstack-sync-upstream-tx/", + toEndsWith: "/profiles/upstream-manifest.json", + message: "injected first apply commit failure", + }], + ); + + assert.notEqual(failed.status, 0); + assert.match(failed.stderr, /injected first apply commit failure/); + assert.deepEqual(snapshotTree(target), before); + assert.equal(existsSync(join(target, "generated")), false); + assertNoTransactionState(target); + + const retried = run(synchronizer, ["--source", source, "--target", target, "--apply"]); + assert.equal(retried.status, 0, retried.stderr); + assert.equal( + readFileSync(join(target, "generated", "deep", "guide", "02-meta-mode.md"), "utf8"), + "Use /meta-mode with mstack.\nHarness confirms.\n", + ); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("dry-run and upstream checks refuse stale state without recovering it", () => { + const { root, source, target } = fixture(); + try { + applyBaseline(source, target); + advanceSource(source, target, () => { + write(join(source, "automations", "benny", "README.md"), "stale pstack automation\n"); + }, "prepare stale read-only checks"); + + const interrupted = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply", "--force"], + root, + [{ + method: "renameSync", + phase: "after", + fromIncludes: "/.mstack-sync-upstream-tx/", + toEndsWith: "/automations/benny/README.md", + action: "exit", + exitCode: 86, + }], + ); + assert.equal(interrupted.status, 86, interrupted.stderr); + const stale = snapshotTree(target); + + const dryRun = run(synchronizer, ["--source", source, "--target", target]); + assert.notEqual(dryRun.status, 0); + assert.match(dryRun.stderr, /active or interrupted upstream sync/i); + assert.deepEqual(snapshotTree(target), stale); + + const checkedWhileStale = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.notEqual(checkedWhileStale.status, 0); + assert.match(checkedWhileStale.stderr, /active or interrupted upstream sync/i); + assert.deepEqual(snapshotTree(target), stale); + + const recovered = run(synchronizer, ["--source", source, "--target", target, "--apply", "--force"]); + assert.equal(recovered.status, 0, recovered.stderr); + assert.equal( + readFileSync(join(target, "automations", "benny", "README.md"), "utf8"), + "stale mstack automation\n", + ); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("rejects a concurrent apply while a live process owns the sync lock", async () => { + const { root, source, target } = fixture(); + const ready = join(root, "lock-ready"); + const release = join(root, "lock-release"); + let first; + try { + const before = snapshotTree(target, { excludeTransactionState: true }); + first = spawnWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply"], + root, + [{ + method: "linkSync", + phase: "after", + toEndsWith: "/.mstack-sync-upstream.lock", + action: "gate", + ready, + release, + }], + ); + await waitForPath(ready); + + const concurrent = runWithFaults( + synchronizer, + ["--source", source, "--target", target, "--apply"], + root, + [], + { timeout: 5000 }, + ); + assert.equal(concurrent.error, undefined, concurrent.error?.message); + assert.notEqual(concurrent.status, 0); + assert.match(concurrent.stderr, /(?:another|active|live).*sync|sync.*(?:lock|active)/i); + assert.deepEqual(snapshotTree(target, { excludeTransactionState: true }), before); + + write(release, "release\n"); + const completed = await first.completed; + assert.equal(completed.status, 0, completed.stderr); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + assertNoTransactionState(target); + } finally { + if (!existsSync(release)) write(release, "release\n"); + if (first?.child.exitCode === null && first?.child.signalCode === null) { + const result = await Promise.race([ + first.completed, + new Promise((resolveWait) => setTimeout(() => resolveWait(undefined), 5000)), + ]); + if (!result) { + first.child.kill(); + await first.completed.catch(() => undefined); + } + } + rmSync(root, { recursive: true, force: true }); + } +}); + +test("takes over dead sync locks and stale takeover guards", () => { + const { root, source, target } = fixture(); + try { + write(join(target, ".mstack-sync-upstream.lock"), `${JSON.stringify({ + version: 1, + pid: 2147483647, + token: "dead-owner", + })}\n`); + write(join(target, ".mstack-sync-upstream.lock.takeover"), `${JSON.stringify({ + version: 1, + pid: 2147483646, + token: "dead-takeover-owner", + })}\n`); + const result = run(synchronizer, ["--source", source, "--target", target, "--apply"]); + assert.equal(result.status, 0, result.stderr); + assertNoTransactionState(target); + const checked = run(checker, ["--source", source, "--target", target, "--strict"]); + assert.equal(checked.status, 0, checked.stderr); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +for (const [kind, description] of [["malformed", "malformed"], ["escaping", "path-escaping"]]) { + test(`refuses a ${description} interrupted transaction journal without mutating files`, () => { + const { root, source, target } = fixture(); + try { + applyBaseline(source, target); + const transactionId = `invalid-${kind}`; + const transaction = transactionDirectory(target, transactionId); + mkdirSync(transaction, { recursive: true }); + write(join(transaction, "COMMITTING"), "\n"); + const outside = join(root, "outside.txt"); + write(outside, "outside remains unchanged\n"); + + if (kind === "malformed") { + write(join(transaction, "journal.json"), "{not json\n"); + } else { + const staged = join(target, ".mstack-sync-upstream-tx", transactionId, "0.new"); + const backup = join(target, ".mstack-sync-upstream-tx", transactionId, "0.old"); + write(staged, "outside replacement\n"); + write(backup, "outside backup\n"); + write(join(transaction, "journal.json"), `${JSON.stringify({ + version: 1, + id: transactionId, + owner: { pid: 2147483647, token: "interrupted-test-owner" }, + targetRoot: target, + createdDirectories: [], + operations: [{ + kind: "write", + target: "../outside.txt", + stage: staged, + backup, + before: { + kind: "file", + sha256: sha256("outside backup\n"), + mode: 0o600, + }, + after: { + kind: "file", + sha256: sha256("outside replacement\n"), + mode: 0o600, + }, + }], + }, null, 2)}\n`); + } + + const before = snapshotTree(target); + const result = run(synchronizer, ["--source", source, "--target", target, "--apply"]); + assert.notEqual(result.status, 0, `${kind} journal was accepted`); + assert.match(result.stderr, /journal|transaction|path|checkout|invalid/i); + assert.deepEqual(snapshotTree(target), before); + assert.equal(readFileSync(outside, "utf8"), "outside remains unchanged\n"); + assert.equal(existsSync(join(target, ".mstack-sync-upstream.lock")), false); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }); +}