From c76e9ec46ebac6a30c8fb05d6b48043b020a8dc9 Mon Sep 17 00:00:00 2001 From: rajasekar-venkatesan <22398308+rajasekar-venkatesan@users.noreply.github.com> Date: Wed, 26 Aug 2026 10:43:51 +0800 Subject: [PATCH 1/9] Fix job records lost on concurrent background task launches updateState() did a lockless read-modify-write of state.json: concurrent `codex-companion.mjs task --background` launches each read the same base state, appended only their own job, and clobbered siblings on write. saveState()'s prune then deleted the orphaned siblings' job files, so most concurrently launched agents vanished from /codex:status and /codex:result. Serialize the read-modify-write with a cross-process O_EXCL lockfile (withStateLock, with jittered backoff and stale-lock stealing) and make the state write atomic (temp file + rename) so lockless readers never observe a torn file. Stress test (12 concurrent upsertJob processes): 2/12 tracked before, 12/12 after. Co-Authored-By: Claude Opus 4.8 --- plugins/codex/scripts/lib/state.mjs | 52 ++++++++++++++++++++++++++--- tests/state.test.mjs | 38 +++++++++++++++++++++ 2 files changed, 86 insertions(+), 4 deletions(-) diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 2da23498f..04c80cc9f 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -111,14 +111,58 @@ export function saveState(cwd, state) { removeFileIfExists(job.logFile); } - fs.writeFileSync(resolveStateFile(cwd), `${JSON.stringify(nextState, null, 2)}\n`, "utf8"); + const stateFile = resolveStateFile(cwd); + const tmpFile = `${stateFile}.${process.pid}.tmp`; + fs.writeFileSync(tmpFile, `${JSON.stringify(nextState, null, 2)}\n`, "utf8"); + fs.renameSync(tmpFile, stateFile); // atomic replace; readers never see a partial file return nextState; } +// Blocking sleep for a sync context (no busy-wait) via Atomics.wait. +function sleepSync(ms) { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); +} + +// Cross-process lock around the state.json read-modify-write. Without it, +// concurrent `task --background` launches each read the same base state, add +// only their own job, and clobber siblings on write (and saveState's prune +// then deletes the "orphan" job files). Serializing the RMW fixes both. +function withStateLock(cwd, fn) { + ensureStateDir(cwd); + const lockFile = path.join(resolveStateDir(cwd), "state.lock"); + const deadline = Date.now() + 15000; + let fd; + for (;;) { + try { + fd = fs.openSync(lockFile, "wx"); // O_CREAT | O_EXCL + break; + } catch (err) { + if (err.code !== "EEXIST") throw err; + // Steal a stale lock left by a crashed process. + try { + if (Date.now() - fs.statSync(lockFile).mtimeMs > 10000) { + fs.unlinkSync(lockFile); + continue; + } + } catch {} + if (Date.now() > deadline) throw new Error("Timed out acquiring Codex state lock"); + sleepSync(20 + Math.floor(Math.random() * 30)); // jittered backoff + } + } + try { + return fn(); + } finally { + try { fs.closeSync(fd); } catch {} + try { fs.unlinkSync(lockFile); } catch {} + } +} + export function updateState(cwd, mutate) { - const state = loadState(cwd); - mutate(state); - return saveState(cwd, state); + return withStateLock(cwd, () => { + const state = loadState(cwd); + mutate(state); + return saveState(cwd, state); + }); } export function generateJobId(prefix = "job") { diff --git a/tests/state.test.mjs b/tests/state.test.mjs index 0f8f57cea..632acaf6d 100644 --- a/tests/state.test.mjs +++ b/tests/state.test.mjs @@ -4,9 +4,15 @@ import path from "node:path"; import test from "node:test"; import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { fileURLToPath, pathToFileURL } from "node:url"; + import { makeTempDir } from "./helpers.mjs"; import { resolveJobFile, resolveJobLogFile, resolveStateDir, resolveStateFile, saveState } from "../plugins/codex/scripts/lib/state.mjs"; +const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const STATE_URL = pathToFileURL(path.join(ROOT, "plugins", "codex", "scripts", "lib", "state.mjs")).href; + test("resolveStateDir uses a temp-backed per-workspace directory", () => { const workspace = makeTempDir(); const stateDir = resolveStateDir(workspace); @@ -103,3 +109,35 @@ test("saveState prunes dropped job artifacts when indexed jobs exceed the cap", .sort() ); }); + +test("updateState serializes concurrent job writes without losing records", async () => { + const workspace = makeTempDir(); + const jobCount = 12; + const worker = + `import { upsertJob } from ${JSON.stringify(STATE_URL)};\n` + + "const [cwd, id] = process.argv.slice(1);\n" + + "upsertJob(cwd, { id, status: \"queued\", jobClass: \"task\", summary: id });\n"; + + await Promise.all( + Array.from({ length: jobCount }, (_, index) => + new Promise((resolve, reject) => { + const child = spawn( + process.execPath, + ["--input-type=module", "-e", worker, workspace, `job-${index}`], + { stdio: "ignore" } + ); + child.on("exit", (code) => + code === 0 ? resolve() : reject(new Error(`worker ${index} exited ${code}`)) + ); + }) + ) + ); + + const saved = JSON.parse(fs.readFileSync(resolveStateFile(workspace), "utf8")); + const trackedIds = new Set(saved.jobs.map((job) => job.id)); + const missing = Array.from({ length: jobCount }, (_, index) => `job-${index}`).filter( + (id) => !trackedIds.has(id) + ); + + assert.deepEqual(missing, [], "every concurrently launched job must be tracked"); +}); From af579cf6db029f463dddd966904280da7c6703a4 Mon Sep 17 00:00:00 2001 From: rajasekar-venkatesan <22398308+rajasekar-venkatesan@users.noreply.github.com> Date: Wed, 26 Aug 2026 12:31:15 +0800 Subject: [PATCH 2/9] Make stale-lock reclaim atomic and ownership-safe Addresses review feedback: the previous stale-lock cleanup did a stat-then-unlink on the lock pathname, which is not atomic. Two racing launches could both observe the same stale lock; after one unlinked it and acquired a fresh lock, the other's unconditional unlink could remove that fresh lock and let a second writer into the read-modify-write concurrently, recreating the lost-job corruption this change prevents. Reclaim the stale lock with rename() instead, which has exactly-one-winner semantics: concurrent reclaimers target the same inode and only one succeeds; losers get ENOENT and fall back to re-contending on open(O_EXCL). Stamp a unique owner id into the lock and only unlink it on release if the contents still match, so a process can never delete a lock another process now holds. Add a regression test that seeds a stale lock and fires concurrent writers, asserting no job records are dropped and the lock is released. --- plugins/codex/scripts/lib/state.mjs | 27 ++++++++++++++----- tests/state.test.mjs | 42 +++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 6 deletions(-) diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 04c80cc9f..54111288a 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -130,18 +130,29 @@ function sleepSync(ms) { function withStateLock(cwd, fn) { ensureStateDir(cwd); const lockFile = path.join(resolveStateDir(cwd), "state.lock"); + // Unique per acquisition: stamped into the lock so only the true owner ever + // removes it, and so a reclaimer's scratch path can't collide. + const ownerId = `${process.pid}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; const deadline = Date.now() + 15000; - let fd; for (;;) { try { - fd = fs.openSync(lockFile, "wx"); // O_CREAT | O_EXCL + const fd = fs.openSync(lockFile, "wx"); // O_CREAT | O_EXCL + fs.writeSync(fd, ownerId); + fs.closeSync(fd); break; } catch (err) { if (err.code !== "EEXIST") throw err; - // Steal a stale lock left by a crashed process. + // Reclaim a stale lock left by a crashed process. Do it atomically: + // rename() has exactly-one-winner semantics, so if several launches race + // to reclaim the same stale lock, only one succeeds and the losers get + // ENOENT and fall back to re-contending on open(O_EXCL). Never unlink the + // lock pathname directly here -- by the time we did, it could already be a + // fresh lock held by another process (the P2 the reviewer flagged). try { if (Date.now() - fs.statSync(lockFile).mtimeMs > 10000) { - fs.unlinkSync(lockFile); + const scratch = `${lockFile}.stale.${ownerId}`; + fs.renameSync(lockFile, scratch); // only one racer wins this + fs.unlinkSync(scratch); continue; } } catch {} @@ -152,8 +163,12 @@ function withStateLock(cwd, fn) { try { return fn(); } finally { - try { fs.closeSync(fd); } catch {} - try { fs.unlinkSync(lockFile); } catch {} + // Only remove the lock if we still own it: if fn() ever outran the stale + // timeout and another process reclaimed the lock, its contents no longer + // match ownerId and we must not delete the lock it now holds. + try { + if (fs.readFileSync(lockFile, "utf8") === ownerId) fs.unlinkSync(lockFile); + } catch {} } } diff --git a/tests/state.test.mjs b/tests/state.test.mjs index 632acaf6d..005ad32db 100644 --- a/tests/state.test.mjs +++ b/tests/state.test.mjs @@ -141,3 +141,45 @@ test("updateState serializes concurrent job writes without losing records", asyn assert.deepEqual(missing, [], "every concurrently launched job must be tracked"); }); + +test("updateState reclaims a stale lock without losing records under concurrency", async () => { + const workspace = makeTempDir(); + + // Simulate a crashed holder: a lock file older than the 10s stale threshold. + const stateDir = resolveStateDir(workspace); + fs.mkdirSync(stateDir, { recursive: true }); + const lockFile = path.join(stateDir, "state.lock"); + fs.writeFileSync(lockFile, "dead-owner"); + const stale = new Date(Date.now() - 60_000); + fs.utimesSync(lockFile, stale, stale); + + const jobCount = 10; + const worker = + `import { upsertJob } from ${JSON.stringify(STATE_URL)};\n` + + "const [cwd, id] = process.argv.slice(1);\n" + + "upsertJob(cwd, { id, status: \"queued\", jobClass: \"task\", summary: id });\n"; + + await Promise.all( + Array.from({ length: jobCount }, (_, index) => + new Promise((resolve, reject) => { + const child = spawn( + process.execPath, + ["--input-type=module", "-e", worker, workspace, `job-${index}`], + { stdio: "ignore" } + ); + child.on("exit", (code) => + code === 0 ? resolve() : reject(new Error(`worker ${index} exited ${code}`)) + ); + }) + ) + ); + + const saved = JSON.parse(fs.readFileSync(resolveStateFile(workspace), "utf8")); + const trackedIds = new Set(saved.jobs.map((job) => job.id)); + const missing = Array.from({ length: jobCount }, (_, index) => `job-${index}`).filter( + (id) => !trackedIds.has(id) + ); + + assert.deepEqual(missing, [], "stale-lock reclaim must not drop concurrent job records"); + assert.equal(fs.existsSync(lockFile), false, "lock file should be released after all writers finish"); +}); From 7d7326a5e9e9af7a3d6ef6f9ab45c6067b46bdbe Mon Sep 17 00:00:00 2001 From: rajasekar-venkatesan <22398308+rajasekar-venkatesan@users.noreply.github.com> Date: Wed, 26 Aug 2026 13:23:10 +0800 Subject: [PATCH 3/9] Make lock reclaim safe by owner liveness and atomic publication Follow-up review found the rename-based reclaim still had a TOCTOU: a delayed reclaimer could act on the lock pathname after another process had recreated a fresh lock there. Reworked the lock so no reclaim decision is ever made against a live owner, and so a lock is never observed in a half-published state. - Reclaim is now gated on owner liveness: the lock stamps its holder's PID and a waiter only reclaims when process.kill(pid, 0) shows the owner is dead. A live owner (even a slow or briefly-suspended one) is never reclaimed, so the removal can never race a lock another process still holds. mtime is demoted to a coarse 10-minute backstop that only covers PID reuse of a dead owner (preserving liveness without letting a live holder be reclaimed in any realistic case). - Locks are published atomically via write-temp + linkSync instead of openSync(O_EXCL) followed by a separate writeSync. linkSync is atomic and fails EEXIST like O_EXCL, but the file carries its full owner id the instant it appears, closing the window where a concurrent reader could see an empty lock and treat a just-created live lock as abandoned. Applied to both the state lock and the serialized reclaim lock. - Reclaim is still serialized through a second lock; that lock is atomically published, PID-stamped, cleared only when its own holder is dead, and released only by its owner. The residual (a holder suspended past the 10-minute backstop, or PID reuse of a dead owner within it) is the timeout limit inherent to any pure-fs PID lock and is documented in the code. --- plugins/codex/scripts/lib/state.mjs | 138 ++++++++++++++++++++++------ tests/state.test.mjs | 7 +- 2 files changed, 113 insertions(+), 32 deletions(-) diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 54111288a..dbbf99581 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -123,6 +123,103 @@ function sleepSync(ms) { Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); } +// Absolute-backstop timeout. A lock is normally reclaimed the instant its owner's +// PID is dead; this only forces reclaim of a lock whose PID still looks alive, to +// preserve liveness in the rare case a dead owner's PID was recycled by an +// unrelated live process. It is deliberately far longer than any real critical +// section (a state.json read-modify-write is milliseconds), so a genuinely live +// holder is never reclaimed by it -- only one wedged/suspended past 10 minutes, +// which is indistinguishable from dead. +const LOCK_BACKSTOP_MS = 600000; + +// Lock files live under a per-workspace OS temp dir (see resolveStateDir), i.e. a +// single host, so a PID read from a lock refers to a process on this machine and +// process.kill(pid, 0) is a valid liveness probe. +function ownerAlive(id) { + const pid = Number.parseInt(String(id).split(".")[0], 10); + if (!Number.isInteger(pid) || pid <= 0) return false; // empty/garbled => not a live owner + try { + process.kill(pid, 0); + return true; + } catch (err) { + return err.code === "EPERM"; // exists but not ours (still alive); ESRCH => dead + } +} + +function isAbandoned(id, mtimeMs) { + if (!ownerAlive(id)) return true; // dead owner -> reclaimable + return Date.now() - mtimeMs > LOCK_BACKSTOP_MS; // else only the far backstop +} + +// Publish a lock atomically: write the owner id into a unique temp file, then +// hard-link it onto the fixed path. linkSync is atomic and fails EEXIST if the +// path is already held, exactly like O_EXCL -- but unlike open()+write() the file +// has its full content the instant it appears at the path, so a concurrent reader +// can never observe an empty lock and mistake a just-created live lock for an +// abandoned one. Returns true if claimed, false if already held. +function claimLock(lockFile, ownerId) { + const tmp = `${lockFile}.tmp.${ownerId}`; + fs.writeFileSync(tmp, ownerId); + try { + fs.linkSync(tmp, lockFile); + return true; + } catch (err) { + if (err.code === "EEXIST") return false; + throw err; + } finally { + try { fs.unlinkSync(tmp); } catch {} + } +} + +// Reclaim a lock only when its owner is gone, never a live one. +// +// Gated on PID liveness: a live owner -- even one working slowly or briefly +// frozen -- keeps a live PID and is never reclaimed, so this can only ever remove +// a lock whose owner is truly dead. A dead owner can neither release nor recreate +// its lock, and no other process can claim the path while it still exists, so +// removing it here cannot race a freshly acquired lock. (The earlier mtime-only +// reclaim was unsafe precisely because it could fire against a live owner and +// unlink a lock another process had recreated in the meantime.) +// +// Removal is serialized through a second reclaim lock so two reclaimers can't +// both act; that lock is atomically published, PID-stamped, only cleared when its +// holder is dead, and released only by its own owner. +function reclaimIfAbandoned(lockFile, reclaimFile, selfId) { + let content, st; + try { + content = fs.readFileSync(lockFile, "utf8"); + st = fs.statSync(lockFile); + } catch { + return; // already gone + } + if (!isAbandoned(content, st.mtimeMs)) return; // live owner -> wait, don't touch + if (!claimLock(reclaimFile, selfId)) { + try { + // A reclaim lock whose own holder died is safe to drop; a live one is left + // alone (its PID is alive, so this never removes a reclaim in progress). + const rc = fs.readFileSync(reclaimFile, "utf8"); + const rst = fs.statSync(reclaimFile); + if (isAbandoned(rc, rst.mtimeMs)) fs.unlinkSync(reclaimFile); + } catch {} + return; + } + try { + // Re-verify under the reclaim lock. The owner is still gone (a dead PID cannot + // come back), and the path can't have been re-claimed while it exists, so + // unlinking here cannot delete a live lock. + const c2 = fs.readFileSync(lockFile, "utf8"); + const s2 = fs.statSync(lockFile); + if (isAbandoned(c2, s2.mtimeMs)) fs.unlinkSync(lockFile); + } catch { + // lock vanished between checks -- fine, nothing to reclaim + } finally { + // Release the reclaim lock only if it is still ours. + try { + if (fs.readFileSync(reclaimFile, "utf8") === selfId) fs.unlinkSync(reclaimFile); + } catch {} + } +} + // Cross-process lock around the state.json read-modify-write. Without it, // concurrent `task --background` launches each read the same base state, add // only their own job, and clobber siblings on write (and saveState's prune @@ -130,42 +227,23 @@ function sleepSync(ms) { function withStateLock(cwd, fn) { ensureStateDir(cwd); const lockFile = path.join(resolveStateDir(cwd), "state.lock"); - // Unique per acquisition: stamped into the lock so only the true owner ever - // removes it, and so a reclaimer's scratch path can't collide. - const ownerId = `${process.pid}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; + const reclaimFile = `${lockFile}.reclaim`; + // ".