fix: keep stale worker-lock recovery exclusive

lib/runtime/process/owner.js

@@ -6,8 +6,32 @@ import { readlinkSync } from 'node:fs';
 import path from 'node:path';
 
 const SHELLS = new Set(['sh', 'bash', 'zsh', 'dash', 'ksh', 'fish', 'cmd.exe', 'powershell', 'pwsh']);
-const PS_TIMEOUT_MS = 1000;
+export const PROCESS_START_TIMEOUT_MS = 1_000;
+// POSIX ps reports start times to the nearest second.
+export const PROCESS_START_TOLERANCE_MS = 2_000;
 const MAX_ANCESTORS = 16;
+const ownStartTime = Math.round(Date.now() - process.uptime() * 1000);
+
+/** A bounded process start-time probe for lock recovery. The hot path needs no subprocess. */
+export function processStartTime(pid, timeout = PROCESS_START_TIMEOUT_MS) {
+  if (pid === process.pid) return ownStartTime;
+  if (!Number.isSafeInteger(pid) || pid <= 0 || timeout <= 0) return null;
+  const windows = process.platform === 'win32';
+  const systemRoot = process.env.SystemRoot ?? 'C:\\Windows';
+  if (windows && !path.isAbsolute(systemRoot)) return null;
+  const executable = windows ? path.join(systemRoot, 'System32', 'WindowsPowerShell', 'v1.0', 'powershell.exe') : '/bin/ps';
+  const args = windows
+    ? ['-NoProfile', '-NonInteractive', '-Command', `([DateTimeOffset] (Get-Process -Id ${pid} -ErrorAction Stop).StartTime).ToUnixTimeMilliseconds()`]
+    : ['-p', String(pid), '-o', 'lstart='];
+  const env = windows ? process.env : { ...process.env, LC_ALL: 'C', TZ: 'UTC' };
+  const result = spawnSync(executable, args, {
+    env, encoding: 'utf8', windowsHide: true, timeout, killSignal: 'SIGKILL',
+  });
+  const started = result.status === 0 ? result.stdout.trim() : '';
+  if (!started) return null;
+  const time = windows ? Number(started) : Date.parse(started + ' GMT');
+  return Number.isSafeInteger(time) && time > 0 ? time : null;
+}
 
 /**
  * Whether the process that wrote a state entry has exited. Signal 0 only probes: ESRCH means no such process;
@@ -35,7 +59,7 @@ function executablePathOf(pid, command) {
 }
 
 export function processInfo(pid) {
-  const probe = spawnSync('ps', ['-o', 'ppid=,comm=', '-p', String(pid)], { encoding: 'utf8', timeout: PS_TIMEOUT_MS });
+  const probe = spawnSync('ps', ['-o', 'ppid=,comm=', '-p', String(pid)], { encoding: 'utf8', timeout: PROCESS_START_TIMEOUT_MS });
   const match = /^\s*(\d+)\s+(.+?)\s*$/.exec(probe.stdout ?? '');
   if (!match) return null;
   const command = match[2].replace(/^-/, '');

lib/runtime/store/admission.js

@@ -0,0 +1,137 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2026 EvoMap
+
+import crypto from 'node:crypto';
+import fs from 'node:fs';
+import path from 'node:path';
+
+import { ownerIsGone, processStartTime, PROCESS_START_TIMEOUT_MS, PROCESS_START_TOLERANCE_MS } from '../process/owner.js';
+
+export const LOCK_WAIT_MS = 2_000;
+export const LOCK_RETRY_MS = 20;
+export const LOCK_ACCESS_ERRORS = new Set(['EPERM', 'EBUSY', 'EACCES']);
+const ADMISSION_IDENTITY_CHECK_MS = 10_000;
+const ABANDONED_PREFIX = 'abandoned-';
+const abandonedAdmissions = new Map();
+
+export function pause(ms) {
+  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms);
+}
+
+function reclaimAdmission(admission, deadline, starts) {
+  try {
+    const info = fs.lstatSync(admission);
+    if (!info.isDirectory()) return false;
+    const entries = fs.readdirSync(admission);
+    if (entries.length > 0) {
+      const marker = entries.find((name) => name.startsWith(ABANDONED_PREFIX));
+      const ownerName = marker ? marker.slice(ABANDONED_PREFIX.length) : entries[0];
+      if (entries.some((name) => name !== ownerName && name !== marker)) return false;
+      const owner = /^owner-(\d+)-[a-f0-9]+(?:-([a-f0-9]{24}|[0-9]{13}))?$/.exec(ownerName);
+      if (!owner) return false;
+      const pid = Number(owner[1]);
+      if (!Number.isSafeInteger(pid) || pid <= 0) return false;
+      const abandonedHere = abandonedAdmissions.get(admission) === ownerName;
+      const markerPath = marker && path.join(admission, marker);
+      if (markerPath && !fs.lstatSync(markerPath).isFile()) return false;
+      if (!markerPath && !abandonedHere && !ownerIsGone(pid)) {
+        const birth = Number(owner[2]);
+        const remaining = deadline - Date.now();
+        if (owner[2]?.length !== 13 || !Number.isSafeInteger(birth) || Date.now() - info.mtimeMs < ADMISSION_IDENTITY_CHECK_MS || remaining <= 0) return false;
+        if (!starts.has(pid)) starts.set(pid, processStartTime(pid, Math.min(remaining, PROCESS_START_TIMEOUT_MS)));
+        const actualBirth = starts.get(pid);
+        // Only a clearly newer process proves reuse. ps timestamps round to whole seconds.
+        if (!actualBirth || actualBirth <= birth + PROCESS_START_TOLERANCE_MS) return false;
+      }
+      if (entries.includes(ownerName)) fs.unlinkSync(path.join(admission, ownerName));
+      if (markerPath) fs.unlinkSync(markerPath);
+      if (abandonedHere) abandonedAdmissions.delete(admission);
+    }
+    // Only the inspected unique owner is unlinked. A fresh guard is already nonempty, so rmdir cannot
+    // remove it even if another reclaimer replaced the directory after our read.
+    fs.rmdirSync(admission);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+function enterAdmission(lockDir, deadline, starts) {
+  const admission = `${lockDir}.admission`;
+  const owner = `owner-${process.pid}-${crypto.randomBytes(12).toString('hex')}-${processStartTime(process.pid)}`;
+  const staging = `${admission}.${owner}.claim`;
+  fs.mkdirSync(path.dirname(admission), { recursive: true, mode: 0o700 });
+  fs.mkdirSync(staging, { mode: 0o700 });
+  let published = false;
+  try {
+    fs.writeFileSync(path.join(staging, owner), '', { mode: 0o600, flag: 'wx' });
+    // Publishing a prepopulated directory never exposes an empty fresh guard to a stale reclaimer.
+    for (let attempt = 0; attempt < 2; attempt += 1) {
+      try {
+        fs.renameSync(staging, admission);
+        published = true;
+        abandonedAdmissions.delete(admission);
+        return { admission, owner };
+      } catch {
+        if (attempt > 0 || !reclaimAdmission(admission, deadline, starts)) return null;
+      }
+    }
+    return null;
+  } finally {
+    if (!published) fs.rmSync(staging, { recursive: true, force: true });
+  }
+}
+
+export function retryLockAccess(access) {
+  for (let attempt = 0; ; attempt += 1) {
+    try {
+      return access();
+    } catch (error) {
+      if (attempt === 2 || !LOCK_ACCESS_ERRORS.has(error?.code)) throw error;
+      pause(LOCK_RETRY_MS);
+    }
+  }
+}
+
+function releaseAdmission({ admission, owner }) {
+  try {
+    retryLockAccess(() => fs.unlinkSync(path.join(admission, owner)));
+  } catch (cause) {
+    if (cause?.code !== 'ENOENT') {
+      throw Object.assign(new Error(`Could not release admission ${admission}`, { cause }), { code: 'ELOCKED' });
+    }
+  }
+  // The guard is released when its unique owner is gone. An empty directory can be reclaimed.
+  try { retryLockAccess(() => fs.rmdirSync(admission)); } catch {}
+}
+
+function abandonAdmission({ admission, owner }) {
+  // A separate marker can be written even when Windows denies access to the owner file.
+  try {
+    retryLockAccess(() => fs.writeFileSync(path.join(admission, ABANDONED_PREFIX + owner), '', { mode: 0o600, flag: 'wx' }));
+  } catch {
+    abandonedAdmissions.set(admission, owner);
+  }
+}
+
+// Every directory mutation uses this short, nonblocking admission. No work or await belongs inside it.
+export function admitted(lockDir, mutate, rollback, deadline = Date.now() + LOCK_WAIT_MS, starts = new Map()) {
+  const admission = enterAdmission(lockDir, deadline, starts);
+  if (!admission) return null;
+  let result;
+  try {
+    result = mutate();
+  } catch (error) {
+    try { releaseAdmission(admission); } catch { abandonAdmission(admission); }
+    throw error;
+  }
+  try {
+    releaseAdmission(admission);
+  } catch (error) {
+    // Publish only after rollback, so another process cannot enter while it runs.
+    try { rollback?.(); } catch (rollbackError) { error.rollbackError = rollbackError; }
+    abandonAdmission(admission);
+    if (rollback) throw error;
+  }
+  return result;
+}

lib/runtime/store/file-lock.js

@@ -5,12 +5,15 @@ import crypto from 'node:crypto';
 import fs from 'node:fs';
 import path from 'node:path';
 
-import { ownerIsGone } from '../process/owner.js';
+import { ownerIsGone, processStartTime, PROCESS_START_TIMEOUT_MS, PROCESS_START_TOLERANCE_MS } from '../process/owner.js';
+
+import { admitted, LOCK_WAIT_MS, LOCK_RETRY_MS, pause, retryLockAccess } from './admission.js';
 
-const LOCK_WAIT_MS = 2_000;
-const LOCK_RETRY_MS = 20;
 const WORK_LOCK_STALE_MS = 2 * 60 * 1000;
+const WORK_OWNER_PROBE_MS = process.platform === 'win32' ? 5_000 : PROCESS_START_TIMEOUT_MS;
+const WORK_OWNER_WITH_START = /^\d+:[a-f0-9]+:([1-9]\d{12})$/;
 const CLAIM_STALE_MS = 60 * 1000;
+const abandonedWork = new Map();
 
 export function readState(target, fallback) {
   try {
@@ -41,11 +44,6 @@ export function removeState(target) {
   }
 }
 
-// Atomics.wait is the only sleep that holds a synchronous read-modify-write together.
-function pause(ms) {
-  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms);
-}
-
 function contentOf(file) {
   try {
     return fs.readFileSync(file, 'utf8');
@@ -179,38 +177,107 @@ function isStale(lockDir, now) {
   }
 }
 
+function workOwnerOf(lockDir) {
+  try {
+    return retryLockAccess(() => fs.readFileSync(path.join(lockDir, 'owner'), 'utf8'));
+  } catch (cause) {
+    if (cause?.code === 'ENOENT') return null;
+    throw Object.assign(new Error(`Could not read lock owner ${lockDir}`, { cause }), { code: 'ELOCKED' });
+  }
+}
+
 function ownsLock(lockDir, token) {
-  return contentOf(path.join(lockDir, 'owner')) === token;
+  return workOwnerOf(lockDir) === token;
 }
 
-function tryLock(lockDir, now) {
-  fs.mkdirSync(path.dirname(lockDir), { recursive: true, mode: 0o700 });
-  if (isStale(lockDir, now)) fs.rmSync(lockDir, { recursive: true, force: true });
-  try {
-    fs.mkdirSync(lockDir);
-  } catch {
-    return null;
-  }
-  const token = crypto.randomBytes(12).toString('hex');
-  fs.writeFileSync(path.join(lockDir, 'owner'), token);
-  return {
-    refresh: () => {
-      if (ownsLock(lockDir, token)) fs.utimesSync(lockDir, new Date(), new Date());
-    },
-    release: () => {
+async function releaseWorkLock(lockDir, token) {
+  const deadline = Date.now() + LOCK_WAIT_MS;
+  const starts = new Map();
+  for (;;) {
+    if (admitted(lockDir, () => {
       if (ownsLock(lockDir, token)) fs.rmSync(lockDir, { recursive: true, force: true });
-    },
-  };
+      return true;
+    }, null, deadline, starts)) return;
+    if (Date.now() >= deadline) throw Object.assign(new Error(`Could not release lock ${lockDir}`), { code: 'ELOCKED' });
+    await new Promise((resolve) => setTimeout(resolve, LOCK_RETRY_MS));
+  }
 }
 
-/** Runs slow `work` while holding a lock directory no other process gets meanwhile; false when it is held. */
+function tryLock(lockDir, now) {
+  let token, suspect, reusedToken;
+  const take = () => admitted(lockDir, () => {
+    const stale = isStale(lockDir, now);
+    if (abandonedWork.has(lockDir) || stale) {
+      let held;
+      try { held = workOwnerOf(lockDir); } catch { return null; }
+      if (held !== abandonedWork.get(lockDir)) {
+        if (!stale) return null;
+        if (held !== reusedToken && /^\d+:/.test(held ?? '')) {
+          const pid = ownerPidOf(held);
+          if (!ownerIsGone(pid)) {
+            const birth = WORK_OWNER_WITH_START.exec(held)?.[1];
+            if (birth) suspect = { token: held, pid, birth: Number(birth) };
+            return null;
+          }
+        }
+      }
+      fs.rmSync(lockDir, { recursive: true, force: true });
+      abandonedWork.delete(lockDir);
+    }
+    try {
+      fs.mkdirSync(lockDir);
+    } catch {
+      return null;
+    }
+    token = `${process.pid}:${crypto.randomBytes(12).toString('hex')}:${processStartTime(process.pid)}`;
+    fs.writeFileSync(path.join(lockDir, 'owner'), token);
+    return {
+      refresh: () => {
+        try {
+          return admitted(lockDir, () => {
+            if (ownsLock(lockDir, token)) fs.utimesSync(lockDir, new Date(), new Date());
+          });
+        } catch {
+          // Timer-driven refresh is best effort; a live pid stays protected and final release still surfaces errors.
+          return null;
+        }
+      },
+      release: async () => {
+        try { await releaseWorkLock(lockDir, token); }
+        catch (error) { abandonedWork.set(lockDir, token); throw error; }
+      },
+    };
+  }, () => {
+    if (!token) return;
+    abandonedWork.set(lockDir, token);
+    if (ownsLock(lockDir, token)) fs.rmSync(lockDir, { recursive: true, force: true });
+    abandonedWork.delete(lockDir);
+  });
+  const lock = take();
+  if (lock || !suspect) return lock;
+  // A foreign lookup can be slow. Release admission, then recheck staleness and the exact token.
+  const actualBirth = processStartTime(suspect.pid, WORK_OWNER_PROBE_MS);
+  if (!actualBirth || actualBirth <= suspect.birth + PROCESS_START_TOLERANCE_MS) return null;
+  reusedToken = suspect.token;
+  return take();
+}
+
+/** Runs slow `work` under one owner's lock; dead or legacy directories recover after their lease expires. */
 export async function withLock(lockDir, work, now = Date.now()) {
   const lock = tryLock(lockDir, now);
   if (!lock) return false;
+  let workFailed = false;
   try {
     await work(lock);
     return true;
+  } catch (error) {
+    workFailed = true;
+    throw error;
   } finally {
-    lock.release();
+    try {
+      await lock.release();
+    } catch (error) {
+      if (!workFailed) throw error;
+    }
   }
 }

test/helpers/lock-contenders.js

@@ -0,0 +1,39 @@
+import { spawn } from 'node:child_process';
+import { rmSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+
+export async function until(check) {
+  const deadline = Date.now() + 10_000;
+  while (!check()) {
+    if (Date.now() >= deadline) throw new Error('timed out waiting for a lock contender');
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+}
+
+export function lockContenders(t, dir, script, args, cleanupMarkers) {
+  const children = [];
+  const results = () => Promise.all(children);
+  t.after(async () => {
+    try {
+      for (const name of cleanupMarkers) writeFileSync(join(dir, name), '');
+    } finally {
+      await results();
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+  return {
+    run(role) {
+      const child = spawn(process.execPath, ['--input-type=module', '-e', script, ...args, role], { windowsHide: true });
+      let stderr = '';
+      const watchdog = setTimeout(() => child.kill('SIGKILL'), 15_000);
+      child.stderr.on('data', (data) => { stderr += data; });
+      child.on('error', (error) => { stderr += String(error); });
+      children.push(new Promise((resolve) => child.on('close', (code) => {
+        clearTimeout(watchdog);
+        resolve({ code, stderr });
+      })));
+      return child;
+    },
+    results,
+  };
+}