fix(skills): preserve replaced checkpoint locks
This commit is contained in:
@@ -38,6 +38,7 @@ const nodeIo = Object.freeze({
|
|||||||
open: (...args) => fs.open(...args),
|
open: (...args) => fs.open(...args),
|
||||||
readFile: (...args) => fs.readFile(...args),
|
readFile: (...args) => fs.readFile(...args),
|
||||||
rename: (...args) => fs.rename(...args),
|
rename: (...args) => fs.rename(...args),
|
||||||
|
stat: (...args) => fs.stat(...args),
|
||||||
unlink: (...args) => fs.unlink(...args),
|
unlink: (...args) => fs.unlink(...args),
|
||||||
randomUUID: () => randomUUID(),
|
randomUUID: () => randomUUID(),
|
||||||
});
|
});
|
||||||
@@ -172,6 +173,10 @@ async function unlinkQuietly(io, path) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function sameFile(left, right) {
|
||||||
|
return left.dev === right.dev && left.ino === right.ino;
|
||||||
|
}
|
||||||
|
|
||||||
async function readExisting(io, target) {
|
async function readExisting(io, target) {
|
||||||
try {
|
try {
|
||||||
return { exists: true, bytes: await io.readFile(target) };
|
return { exists: true, bytes: await io.readFile(target) };
|
||||||
@@ -260,6 +265,7 @@ export async function checkpointRecord(request) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let lockHandle = null;
|
let lockHandle = null;
|
||||||
|
let lockIdentity = null;
|
||||||
let tempHandle = null;
|
let tempHandle = null;
|
||||||
let tempPath = null;
|
let tempPath = null;
|
||||||
let primaryError;
|
let primaryError;
|
||||||
@@ -267,13 +273,14 @@ export async function checkpointRecord(request) {
|
|||||||
try {
|
try {
|
||||||
try {
|
try {
|
||||||
lockHandle = await io.open(lock, "wx", FILE_MODE);
|
lockHandle = await io.open(lock, "wx", FILE_MODE);
|
||||||
|
lockIdentity = await lockHandle.stat();
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
if (isAlreadyExists(error)) {
|
if (isAlreadyExists(error)) {
|
||||||
throw new CheckpointError("LOCK_EXISTS", "checkpoint target is locked; refusing to recover or overwrite the lock", {
|
throw new CheckpointError("LOCK_EXISTS", "checkpoint target is locked; refusing to recover or overwrite the lock", {
|
||||||
cause: error,
|
cause: error,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
throw new CheckpointError("LOCK_FAILED", `could not acquire checkpoint lock: ${messageFor(error)}`, {
|
throw new CheckpointError("LOCK_FAILED", `could not acquire and identify checkpoint lock: ${messageFor(error)}`, {
|
||||||
cause: error,
|
cause: error,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -369,12 +376,24 @@ export async function checkpointRecord(request) {
|
|||||||
cause: lockCloseError,
|
cause: lockCloseError,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
if (lockHandle !== null) {
|
if (lockIdentity !== null) {
|
||||||
const lockUnlinkError = await unlinkQuietly(io, lock);
|
let currentLockIdentity;
|
||||||
if (lockUnlinkError !== undefined && !isMissing(lockUnlinkError) && primaryError === undefined) {
|
try {
|
||||||
primaryError = new CheckpointError("LOCK_CLEANUP_FAILED", `could not remove checkpoint lock: ${messageFor(lockUnlinkError)}`, {
|
currentLockIdentity = await io.stat(lock);
|
||||||
cause: lockUnlinkError,
|
} catch (error) {
|
||||||
});
|
if (!isMissing(error) && primaryError === undefined) {
|
||||||
|
primaryError = new CheckpointError("LOCK_CLEANUP_FAILED", `could not verify checkpoint lock ownership: ${messageFor(error)}`, {
|
||||||
|
cause: error,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (currentLockIdentity !== undefined && sameFile(lockIdentity, currentLockIdentity)) {
|
||||||
|
const lockUnlinkError = await unlinkQuietly(io, lock);
|
||||||
|
if (lockUnlinkError !== undefined && !isMissing(lockUnlinkError) && primaryError === undefined) {
|
||||||
|
primaryError = new CheckpointError("LOCK_CLEANUP_FAILED", `could not remove checkpoint lock: ${messageFor(lockUnlinkError)}`, {
|
||||||
|
cause: lockUnlinkError,
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -175,6 +175,36 @@ test("serializes concurrent updates with an exclusive per-target lock", async ()
|
|||||||
assert.equal(await exists(`${target}.lock`), false);
|
assert.equal(await exists(`${target}.lock`), false);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("does not remove a lock pathname replaced before cleanup", async () => {
|
||||||
|
const directory = await temporaryDirectory();
|
||||||
|
const target = join(directory, "record.md");
|
||||||
|
const lock = `${target}.lock`;
|
||||||
|
const replacement = "replacement lock";
|
||||||
|
let replaced = false;
|
||||||
|
const io = {
|
||||||
|
...defaultIo,
|
||||||
|
open: async (path, flags, mode) => {
|
||||||
|
const handle = await defaultIo.open(path, flags, mode);
|
||||||
|
if (path !== lock || flags !== "wx") return handle;
|
||||||
|
return {
|
||||||
|
stat: (...args) => handle.stat(...args),
|
||||||
|
close: async () => {
|
||||||
|
await handle.close();
|
||||||
|
await fs.unlink(lock);
|
||||||
|
await fs.writeFile(lock, replacement, { mode: 0o600 });
|
||||||
|
replaced = true;
|
||||||
|
},
|
||||||
|
};
|
||||||
|
},
|
||||||
|
};
|
||||||
|
|
||||||
|
await checkpointRecord({ target, input: record(), io });
|
||||||
|
|
||||||
|
assert.equal(replaced, true);
|
||||||
|
assert.equal(await fs.readFile(lock, "utf8"), replacement);
|
||||||
|
assert.equal(await fs.readFile(target, "utf8"), record());
|
||||||
|
});
|
||||||
|
|
||||||
test("fails closed on a pre-existing stale lock and leaves it for recovery", async () => {
|
test("fails closed on a pre-existing stale lock and leaves it for recovery", async () => {
|
||||||
const directory = await temporaryDirectory();
|
const directory = await temporaryDirectory();
|
||||||
const target = join(directory, "record.md");
|
const target = join(directory, "record.md");
|
||||||
|
|||||||
Reference in New Issue
Block a user