stack/packages/queue/tests/lock.test.mjs

// 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), []);
});

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 }), /^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, /^removed queue lock \(dead/);
  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.equal(out, "no queue lock present; nothing removed");
  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.match(out, /^no queue lock present; nothing removed\nwarning: 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 }), /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 }), /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.equal(unlock({ gitDir: d, io: realIo }), "no queue lock present; nothing removed");
  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);
});