// The lock and the unlock gate (8.4): each schedule in the 8.4 test list. import assert from "node:assert/strict"; import { spawn } from "node:child_process"; import { existsSync, mkdtempSync, readdirSync, readFileSync, realpathSync, renameSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { test } from "node:test"; import { bootId, processStart } from "../../discord/src/journal.mjs"; import { QueueError } from "../src/errors.mjs"; import { realIo } from "../src/io.mjs"; import { GATE_NAME, LOCK_NAME, acquire, checkGate, classify, realProc, release, unlock } from "../src/lock.mjs"; const HERE = new URL(".", import.meta.url).pathname; function dir(t) { const d = realpathSync(mkdtempSync(join(tmpdir(), "mosaic-queue-lock-"))); t.after(() => rmSync(d, { recursive: true, force: true })); return d; } const OTHER_BOOT = "00000000-0000-4000-8000-000000000000"; function record(over = {}) { return Buffer.from(JSON.stringify({ pid: process.pid, start: processStart(process.pid), boot: bootId(), host: realProc.host(), op: "some-op-0001", verb: "move", at: "2026-09-26T00:00:00.000Z", ...over, }) + "\n"); } // A pid that is not running: a child that has already exited. async function deadPid() { const c = spawn(process.execPath, ["-e", ""]); await new Promise((r) => c.on("exit", r)); return c.pid; } // A live process that the tests never signal except to clean up. async function sleeper(t) { const c = spawn("sleep", ["60"], { stdio: "ignore" }); await new Promise((r) => c.on("spawn", r)); t.after(() => { try { c.kill("SIGKILL"); } catch { /* gone */ } }); return c; } function refused(fn, re, code = 2) { assert.throws(fn, (err) => err instanceof QueueError && err.code === code && re.test(err.message)); } const tmps = (d) => readdirSync(d).filter((n) => n.endsWith(".tmp")); test("acquire publishes the record by link; release removes only its own lock", (t) => { const d = dir(t); const h = acquire({ gitDir: d, io: realIo, op: "abcdefgh-1", verb: "move" }); const rec = JSON.parse(readFileSync(join(d, LOCK_NAME), "utf8")); assert.deepEqual(Object.keys(rec), ["pid", "start", "boot", "host", "op", "verb", "at"]); assert.equal(classify(readFileSync(join(d, LOCK_NAME)), realProc).state, "live"); assert.deepEqual(tmps(d), []); assert.equal(release(h, realIo), null); assert.equal(existsSync(join(d, LOCK_NAME)), false); }); test("a kill between the temp write and the link leaves no lock", async (t) => { const d = dir(t); const child = spawn(process.execPath, [join(HERE, "fixtures", "lock-child.mjs"), d, "kill-before-link"], { stdio: "ignore" }); const [, signal] = await new Promise((r) => child.on("exit", (...a) => r(a))); assert.equal(signal, "SIGKILL"); assert.equal(existsSync(join(d, LOCK_NAME)), false); assert.equal(tmps(d).length, 1, "the killed writer's temp file stays; it is not the lock"); release(acquire({ gitDir: d, io: realIo, verb: "move" }), realIo); }); test("a short or failed temp write refuses and leaves no lock and no temp", (t) => { const d = dir(t); const short = { ...realIo, write: () => 0 }; refused(() => acquire({ gitDir: d, io: short, verb: "move" }), /cannot write the lock record .*ESHORT/, 1); const enospc = { ...realIo, write: () => { throw Object.assign(new Error("full"), { code: "ENOSPC" }); } }; refused(() => acquire({ gitDir: d, io: enospc, verb: "move" }), /ENOSPC; no lock taken/, 1); const badFsync = { ...realIo, fsync: () => { throw Object.assign(new Error("io"), { code: "EIO" }); } }; refused(() => acquire({ gitDir: d, io: badFsync, verb: "move" }), /EIO/, 1); const badBack = { ...realIo, readFile: (p) => (p.endsWith(".tmp") ? Buffer.from("x") : realIo.readFile(p)) }; refused(() => acquire({ gitDir: d, io: badBack, verb: "move" }), /EREADBACK/, 1); assert.deepEqual(readdirSync(d), []); }); test("a link error other than EEXIST refuses", (t) => { const d = dir(t); const io = { ...realIo, link: () => { throw Object.assign(new Error("no"), { code: "EPERM" }); } }; refused(() => acquire({ gitDir: d, io, verb: "move" }), /cannot link .*EPERM; no lock taken/, 1); assert.deepEqual(readdirSync(d), []); }); test("an error after the link releases the lock: unreadable gate, failing temp stat", (t) => { const d = dir(t); writeFileSync(join(d, GATE_NAME), record({ verb: "unlock", op: null })); const denied = { ...realIo, readFile: (p) => (p.endsWith(GATE_NAME) ? (() => { throw Object.assign(new Error("denied"), { code: "EACCES" }); })() : realIo.readFile(p)) }; refused(() => acquire({ gitDir: d, io: denied, verb: "move" }), /cannot check the unlock gate .*EACCES; lock released$/, 1); assert.deepEqual(readdirSync(d), [GATE_NAME]); rmSync(join(d, GATE_NAME)); const badStat = { ...realIo, stat: () => { throw Object.assign(new Error("io"), { code: "EIO" }); } }; refused(() => acquire({ gitDir: d, io: badStat, verb: "move" }), /cannot write the lock record .*EIO; no lock taken/, 1); assert.deepEqual(readdirSync(d), []); // A stat that fails from its second call on: publish stats once, before the // link, so nothing after the link can fail and strand the lock. let stats = 0; const lateStat = { ...realIo, stat: (p) => { if (++stats > 1) throw Object.assign(new Error("io"), { code: "EIO" }); return realIo.stat(p); } }; const h = acquire({ gitDir: d, io: lateStat, verb: "move" }); assert.equal(stats, 1); assert.equal(release(h, realIo), null); assert.deepEqual(readdirSync(d), []); }); // P1: a release that throws on either gate path becomes a message naming the // lock left behind, and the gate-present refusal reports what release found. test("a release that fails on a gate path is reported, never a stack trace (P1)", (t) => { const d = dir(t); const lock = join(d, LOCK_NAME); const eacces = () => { throw Object.assign(new Error("denied"), { code: "EACCES" }); }; const noUnlink = { ...realIo, unlink: (p) => (p === lock ? eacces() : realIo.unlink(p)) }; // The gate is present. writeFileSync(join(d, GATE_NAME), record({ verb: "unlock", op: null })); refused(() => acquire({ gitDir: d, io: noUnlink, verb: "move" }), /unlock gate .* is present \(live: .*--check-gate`\nwarning: cannot release the queue lock \(EACCES\); .*mosaic-queue\.lock may be left in place$/); assert.equal(existsSync(lock), true); rmSync(lock); // The gate can't be read. const both = { ...noUnlink, readFile: (p) => (p.endsWith(GATE_NAME) ? eacces() : realIo.readFile(p)) }; refused(() => acquire({ gitDir: d, io: both, verb: "move" }), /cannot check the unlock gate .*EACCES; cannot release the queue lock \(EACCES\); .*mosaic-queue\.lock may be left in place$/, 1); assert.equal(existsSync(lock), true); rmSync(lock); // The gate is present and the lock was swapped: release's own message is kept. const swap = (name) => { if (name === "lock-linked") { writeFileSync(`${lock}.copy`, readFileSync(lock)); renameSync(`${lock}.copy`, lock); } }; refused(() => acquire({ gitDir: d, io: realIo, verb: "move", hook: swap }), /is present \(live: .*\nwarning: lock .*mosaic-queue\.lock is not the one this process took; left in place$/); assert.equal(existsSync(lock), true); rmSync(lock); // Nothing wrong: the refusal carries no warning. refused(() => acquire({ gitDir: d, io: realIo, verb: "move" }), /--check-gate`$/); assert.deepEqual(readdirSync(d), [GATE_NAME]); }); test("a paused holder: another writer waits 10 s, then refuses naming it live", async (t) => { const d = dir(t); const child = spawn(process.execPath, [join(HERE, "fixtures", "lock-child.mjs"), d, "hold"], { stdio: ["ignore", "pipe", "ignore"] }); t.after(() => { try { child.kill("SIGKILL"); } catch { /* gone */ } }); await new Promise((r) => child.stdout.once("data", r)); const t0 = Date.now(); refused(() => acquire({ gitDir: d, io: realIo, op: "waiter-op-1", verb: "move" }), /queue lock held by move holder-op-1 since .*; retry the same op later/); assert.ok(Date.now() - t0 >= 10000, `waited ${Date.now() - t0} ms`); child.kill("SIGKILL"); await new Promise((r) => child.on("exit", r)); refused(() => acquire({ gitDir: d, io: realIo, verb: "move", waitMs: 0 }), /owner is dead: pid .*; run `scripts\/mosaic queue unlock` once nothing is running/); assert.match(unlock({ gitDir: d, io: realIo }).result, /^removed queue lock \(dead: pid/); release(acquire({ gitDir: d, io: realIo, verb: "move", waitMs: 0 }), realIo); }); test("two concurrent unlockers: the second refuses on the gate", async (t) => { const d = dir(t); writeFileSync(join(d, LOCK_NAME), record({ pid: await deadPid() })); let inner; const outer = unlock({ gitDir: d, io: realIo, hook: (name) => { if (name !== "gate-held") return; try { unlock({ gitDir: d, io: realIo }); } catch (err) { inner = err; } }, }); assert.match(inner.message, /unlock gate .* is held \(live: pid \d+, unlock since/); assert.match(outer.result, /^removed queue lock \(dead/); assert.equal(outer.warning, null); assert.equal(existsSync(join(d, GATE_NAME)), false); }); test("a writer publishing during an unlock, lock first: unlock sees it live and refuses", (t) => { const d = dir(t); let unlockErr; const h = acquire({ gitDir: d, io: realIo, verb: "move", hook: (name) => { if (name !== "lock-linked") return; try { unlock({ gitDir: d, io: realIo }); } catch (err) { unlockErr = err; } }, }); assert.match(unlockErr.message, /queue lock owner is live: .*; unlock refuses/); assert.equal(existsSync(join(d, GATE_NAME)), false); assert.equal(release(h, realIo), null); }); test("a writer publishing during an unlock, gate first: the writer releases and refuses", (t) => { const d = dir(t); let writerErr; const out = unlock({ gitDir: d, io: realIo, hook: (name) => { if (name !== "gate-held") return; try { acquire({ gitDir: d, io: realIo, verb: "move" }); } catch (err) { writerErr = err; } }, }); assert.match(writerErr.message, /unlock gate .* is present \(live: .*unlock/); assert.deepEqual(out, { result: "no queue lock present; nothing removed", warning: null }); assert.deepEqual(readdirSync(d), []); }); test("a gate swapped while held is left in place and reported, on success and on refusal (N1)", (t) => { const d = dir(t); const gate = join(d, GATE_NAME); // A copy renamed over the gate: same bytes, a new inode. const swap = () => { writeFileSync(`${gate}.copy`, readFileSync(gate)); renameSync(`${gate}.copy`, gate); }; const out = unlock({ gitDir: d, io: realIo, hook: (name) => { if (name === "gate-held") swap(); } }); assert.equal(out.result, "no queue lock present; nothing removed"); assert.match(out.warning, /^lock .*mosaic-queue\.unlock is not the one this process took; left in place$/); assert.equal(existsSync(gate), true); rmSync(gate); writeFileSync(join(d, LOCK_NAME), record({})); refused(() => unlock({ gitDir: d, io: realIo, hook: (name) => { if (name === "gate-held") swap(); } }), /owner is live: .*; unlock refuses\nwarning: lock .*mosaic-queue\.unlock is not the one this process took; left in place$/); assert.equal(existsSync(gate), true); assert.equal(existsSync(join(d, LOCK_NAME)), true); }); test("a reused pid within one boot is mismatch; unlock removes the lock and never signals the process", async (t) => { const d = dir(t); const s = await sleeper(t); const start = processStart(s.pid); const wrong = String(BigInt(start) + 1n); writeFileSync(join(d, LOCK_NAME), record({ pid: s.pid, start: wrong })); assert.equal(classify(readFileSync(join(d, LOCK_NAME)), realProc).state, "mismatch"); refused(() => acquire({ gitDir: d, io: realIo, verb: "move", waitMs: 0 }), /mismatch: .*was reused/); assert.match(unlock({ gitDir: d, io: realIo }).result, /removed queue lock \(mismatch/); assert.equal(s.exitCode, null); assert.equal(processStart(s.pid), start, "the process still runs, unsignalled"); }); test("the same pid and start on a different boot is mismatch", (t) => { const d = dir(t); writeFileSync(join(d, LOCK_NAME), record({ boot: OTHER_BOOT })); const c = classify(readFileSync(join(d, LOCK_NAME)), realProc); assert.deepEqual([c.state, c.why], ["mismatch", "recorded in a previous boot"]); }); test("a foreign host is unknown whatever the local pid says; unlock refuses", async (t) => { const d = dir(t); writeFileSync(join(d, LOCK_NAME), record({ pid: await deadPid(), host: "some-other-host" })); assert.equal(classify(readFileSync(join(d, LOCK_NAME)), realProc).state, "unknown"); refused(() => acquire({ gitDir: d, io: realIo, verb: "move", waitMs: 0 }), /unknown: .*recorded on host some-other-host; unlock refuses this too/); refused(() => unlock({ gitDir: d, io: realIo }), /owner is unknown: .*; unlock refuses/); assert.equal(existsSync(join(d, LOCK_NAME)), true); // A real foreign host has its own boot id. Host is tested before boot, so // this is still unknown, never mismatch, and unlock still refuses. writeFileSync(join(d, LOCK_NAME), record({ pid: await deadPid(), host: "some-other-host", boot: OTHER_BOOT })); assert.equal(classify(readFileSync(join(d, LOCK_NAME)), realProc).state, "unknown"); refused(() => unlock({ gitDir: d, io: realIo }), /owner is unknown: .*recorded on host some-other-host/); assert.equal(existsSync(join(d, LOCK_NAME)), true); }); test("unreadable /proc: classification is unknown and acquire refuses", (t) => { const d = dir(t); const blind = { ...realProc, processStart: () => null }; assert.equal(classify(record(), blind).state, "unknown"); const alive = { ...realProc, processStart: (pid) => (pid === process.pid ? realProc.processStart(pid) : null) }; assert.match(classify(record({ pid: 1 }), alive).why, /start time is unreadable/); refused(() => acquire({ gitDir: d, io: realIo, proc: blind, verb: "move" }), /cannot read this process's start time or boot id/); refused(() => acquire({ gitDir: d, io: realIo, proc: { ...realProc, bootId: () => null }, verb: "move" }), /cannot read/); }); test("invalid records: empty, unparsable, wrong keys, bad start or boot", () => { for (const b of [null, "", "{", "[]", JSON.stringify({ pid: 1 }), record({ start: "x" }).toString(), record({ boot: "nope" }).toString()]) { assert.equal(classify(b === null ? null : Buffer.from(b), realProc).state, "invalid"); } }); test("a stale gate blocks writers; --check-gate says mismatch for a reused pid", async (t) => { const d = dir(t); const s = await sleeper(t); writeFileSync(join(d, GATE_NAME), record({ pid: s.pid, start: String(BigInt(processStart(s.pid)) + 1n), verb: "unlock", op: null })); refused(() => acquire({ gitDir: d, io: realIo, verb: "move" }), /unlock gate .* is present \(mismatch: .*--check-gate/); assert.equal(existsSync(join(d, LOCK_NAME)), false, "the writer released its lock"); const g = checkGate({ gitDir: d, io: realIo }); assert.equal(g.state, "mismatch"); assert.match(g.line, /remove .* by hand only once no queue command is running/); refused(() => unlock({ gitDir: d, io: realIo }), /unlock gate .* is held \(mismatch/); writeFileSync(join(d, GATE_NAME), record({ host: "elsewhere", verb: "unlock", op: null })); assert.match(checkGate({ gitDir: d, io: realIo }).line, /unknown: .*leave it for diagnosis/); rmSync(join(d, GATE_NAME)); assert.equal(checkGate({ gitDir: d, io: realIo }).state, "absent"); }); test("a delayed release by a dead owner, after unlock and a new owner: the inode check keeps the new lock", (t) => { const d = dir(t); const a = acquire({ gitDir: d, io: realIo, op: "owner-a-op", verb: "move" }); // Unlock judges A dead (as it would after a crash); B then takes the lock. const judge = { ...realProc, pidAlive: () => false }; assert.match(unlock({ gitDir: d, io: realIo, proc: judge }).result, /removed queue lock \(dead/); const b = acquire({ gitDir: d, io: realIo, op: "owner-b-op", verb: "move" }); assert.match(release(a, realIo), /is not the one this process took; left in place/); assert.match(readFileSync(join(d, LOCK_NAME), "utf8"), /owner-b-op/); assert.equal(release(b, realIo), null); assert.match(release(b, realIo), /already gone/); }); test("release checks the inode too: a byte-identical lock file with a new inode is left in place", (t) => { const d = dir(t); const a = acquire({ gitDir: d, io: realIo, op: "owner-a-op", verb: "move" }); const bytes = readFileSync(join(d, LOCK_NAME)); // The copy exists alongside the original before the rename, so its inode differs. writeFileSync(join(d, "copy"), bytes); renameSync(join(d, "copy"), join(d, LOCK_NAME)); assert.match(release(a, realIo), /is not the one this process took; left in place/); assert.deepEqual(readFileSync(join(d, LOCK_NAME)), bytes); }); test("unlock refuses a live, unknown or invalid lock, and does nothing without one", async (t) => { const d = dir(t); assert.deepEqual(unlock({ gitDir: d, io: realIo }), { result: "no queue lock present; nothing removed", warning: null }); writeFileSync(join(d, LOCK_NAME), ""); refused(() => unlock({ gitDir: d, io: realIo }), /owner is invalid/); writeFileSync(join(d, LOCK_NAME), record()); refused(() => unlock({ gitDir: d, io: realIo }), /owner is live/); refused(() => acquire({ gitDir: d, io: realIo, verb: "move", waitMs: 0 }), /held by move some-op-0001/); writeFileSync(join(d, LOCK_NAME), "not json"); refused(() => acquire({ gitDir: d, io: realIo, verb: "move", waitMs: 0 }), /record is invalid; inspect .* by hand/); assert.equal(existsSync(join(d, GATE_NAME)), false); });