From 9e816dae23b93b2744afaec656299fab9b30a7b9 Mon Sep 17 00:00:00 2001 From: GautamSharma99 Date: Thu, 30 Jul 2026 02:17:25 +0530 Subject: [PATCH 1/2] fix: publish multiscan locks atomically (cherry picked from commit f6e30339a816cff8a2a01b909032d7e51df74125) --- sdk/typescript/src/multiscan.ts | 147 +++++++++++++++++++--- sdk/typescript/tests-ts/multiscan.test.ts | 49 +++++++- 2 files changed, 176 insertions(+), 20 deletions(-) diff --git a/sdk/typescript/src/multiscan.ts b/sdk/typescript/src/multiscan.ts index 534715a6..86a7e20c 100644 --- a/sdk/typescript/src/multiscan.ts +++ b/sdk/typescript/src/multiscan.ts @@ -1,6 +1,7 @@ import { execFile as execFileCallback } from "node:child_process"; import { randomUUID } from "node:crypto"; import { + link, lstat, mkdir, open, @@ -28,6 +29,12 @@ const REQUIRED_ARTIFACTS = [ "coverage.json", "report.md", ]; +const MAX_LOCK_OWNER_BYTES = 4 * 1024; + +interface MultiscanLockOwner { + pid: number; + token?: string; +} interface MultiscanTask { id: string; @@ -267,30 +274,132 @@ async function appendReceipt(path: string, receipt: string): Promise { async function acquireLock(output: string): Promise<() => Promise> { const path = join(output, ".lock"); - const ownerPath = join(path, "owner.json"); + const token = randomUUID(); + const pending = join(output, `.lock.pending-${token}`); + const owner = `${JSON.stringify({ pid: process.pid, token })}\n`; + const file = await open(pending, "wx", 0o600); + try { + await file.writeFile(owner, "utf8"); + await file.sync(); + } finally { + await file.close(); + } + + try { + for (;;) { + try { + await link(pending, path); + return async () => await releaseLock(path, token); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + } + + const existing = await readLockOwner(path); + if (existing !== null && processIsRunning(existing.pid)) { + const current = await readLockOwner(path); + if ( + current !== null && + current.pid === existing.pid && + current.token === existing.token + ) { + throw new Error("A multiscan supervisor is already running."); + } + continue; + } + + const stale = join(output, `.lock.stale-${randomUUID()}`); + try { + await rename(path, stale); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw error; + } + const moved = await readLockOwner(stale); + if (moved !== null && processIsRunning(moved.pid)) { + try { + await rename(stale, path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + } + throw new Error("A multiscan supervisor is already running."); + } + await rm(stale, { recursive: true, force: true }); + } + } finally { + await rm(pending, { force: true }); + } +} + +async function readLockOwner(path: string): Promise { + let metadata; try { - await mkdir(path, { mode: 0o700 }); + metadata = await lstat(path); } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; - const { pid } = JSON.parse(await readFile(ownerPath, "utf8")) as { - pid: number; - }; + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + } + let ownerPath = path; + if (metadata.isDirectory()) { + ownerPath = join(path, "owner.json"); try { - process.kill(pid, 0); - throw new Error("A multiscan supervisor is already running."); - } catch (failure) { - if ((failure as NodeJS.ErrnoException).code !== "ESRCH") throw failure; + metadata = await lstat(ownerPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; } - const stale = join(output, `.lock.stale-${randomUUID()}`); - await rename(path, stale); - await rm(stale, { recursive: true }); - return await acquireLock(output); } - await writeFile(ownerPath, `${JSON.stringify({ pid: process.pid })}\n`, { - flag: "wx", - mode: 0o600, - }); - return async () => rm(path, { recursive: true }); + if ( + !metadata.isFile() || + metadata.isSymbolicLink() || + metadata.size > MAX_LOCK_OWNER_BYTES + ) { + return null; + } + let value: unknown; + try { + value = JSON.parse(await readFile(ownerPath, "utf8")); + } catch (error) { + if ( + error instanceof SyntaxError || + (error as NodeJS.ErrnoException).code === "ENOENT" + ) { + return null; + } + throw error; + } + if ( + typeof value !== "object" || + value === null || + !("pid" in value) || + !Number.isSafeInteger(value.pid) || + (value.pid as number) <= 0 || + ("token" in value && typeof value.token !== "string") + ) { + return null; + } + return { + pid: value.pid as number, + ...("token" in value ? { token: value.token as string } : {}), + }; +} + +function processIsRunning(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === "ESRCH") return false; + if (code === "EPERM") return true; + throw error; + } +} + +async function releaseLock(path: string, token: string): Promise { + const owner = await readLockOwner(path); + if (owner?.token === token) { + await rm(path, { force: true }); + } } async function ensureManifest( diff --git a/sdk/typescript/tests-ts/multiscan.test.ts b/sdk/typescript/tests-ts/multiscan.test.ts index f357fac9..d08878cf 100644 --- a/sdk/typescript/tests-ts/multiscan.test.ts +++ b/sdk/typescript/tests-ts/multiscan.test.ts @@ -2,6 +2,7 @@ import { execFileSync } from "node:child_process"; import { access, appendFile, + lstat, mkdir, mkdtemp, readFile, @@ -412,9 +413,20 @@ describe("multiscan", () => { await finish; return await completedScan(scanOptions.outputDir!); }); + const lock = join(paths.output, ".lock"); const first = runMultiscan(options(paths, security)); await running; try { + expect((await lstat(lock)).isFile()).toBe(true); + expect(JSON.parse(await readFile(lock, "utf8"))).toMatchObject({ + pid: process.pid, + token: expect.any(String), + }); + expect( + (await readdir(paths.output)).some((name) => + name.startsWith(".lock.pending-"), + ), + ).toBe(false); await expect(runMultiscan(options(paths, security))).rejects.toThrow( /running|locked|supervisor/iu, ); @@ -425,7 +437,6 @@ describe("multiscan", () => { const [receipt] = await results(join(paths.output, "results.jsonl")); await rm(join(receipt!["outputDir"] as string, "report.md")); - const lock = join(paths.output, ".lock"); await mkdir(lock); await writeFile( join(lock, "owner.json"), @@ -440,6 +451,42 @@ describe("multiscan", () => { await expect(access(lock)).rejects.toThrow(); }); + test("recovers missing and malformed legacy lock ownership", async () => { + const paths = await fixture(); + const source = await repository(paths.root, "lock-recovery"); + await writeFile( + paths.input, + `id,repository,revision\nlock-recovery,${source.path},${source.revision}\n`, + ); + await mkdir(paths.output); + const lock = join(paths.output, ".lock"); + let scans = 0; + const security = client(async (_repository, scanOptions = {}) => { + scans += 1; + return await completedScan(scanOptions.outputDir!); + }); + + for (const owner of [undefined, '{"pid":', '{"pid":"invalid"}']) { + await mkdir(lock); + if (owner !== undefined) { + await writeFile(join(lock, "owner.json"), owner); + } + expect(await runMultiscan(options(paths, security))).toMatchObject({ + completed: 1, + failed: 0, + }); + await expect(access(lock)).rejects.toThrow(); + } + + expect(scans).toBe(1); + expect( + (await readdir(paths.output)).some( + (name) => + name.startsWith(".lock.pending-") || name.startsWith(".lock.stale-"), + ), + ).toBe(false); + }); + test("retries a failed attempt and records both durable receipts", async () => { const paths = await fixture(); const source = await repository(paths.root, "retry"); From 8cdc1f68809a9a0c3a22098dee465e9e90f29638 Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Sat, 1 Aug 2026 06:51:23 -0700 Subject: [PATCH 2/2] fix: serialize multiscan lock recovery and preserve ownership --- sdk/typescript/src/multiscan.ts | 94 +++++--- sdk/typescript/tests-ts/multiscan.test.ts | 263 +++++++++++++++++++++- 2 files changed, 329 insertions(+), 28 deletions(-) diff --git a/sdk/typescript/src/multiscan.ts b/sdk/typescript/src/multiscan.ts index 86a7e20c..c785d6c1 100644 --- a/sdk/typescript/src/multiscan.ts +++ b/sdk/typescript/src/multiscan.ts @@ -30,6 +30,7 @@ const REQUIRED_ARTIFACTS = [ "report.md", ]; const MAX_LOCK_OWNER_BYTES = 4 * 1024; +const MAX_LOCK_OWNER_PID = 2_147_483_647; interface MultiscanLockOwner { pid: number; @@ -274,21 +275,31 @@ async function appendReceipt(path: string, receipt: string): Promise { async function acquireLock(output: string): Promise<() => Promise> { const path = join(output, ".lock"); + const recovery = join(output, ".lock.recovery"); const token = randomUUID(); const pending = join(output, `.lock.pending-${token}`); const owner = `${JSON.stringify({ pid: process.pid, token })}\n`; const file = await open(pending, "wx", 0o600); try { - await file.writeFile(owner, "utf8"); - await file.sync(); - } finally { - await file.close(); - } + try { + await file.writeFile(owner, "utf8"); + await file.sync(); + } finally { + await file.close(); + } - try { for (;;) { try { await link(pending, path); + const recovering = await readLockOwner(recovery); + if ( + recovering !== null && + recovering.token !== token && + processIsRunning(recovering.pid) + ) { + await releaseLock(path, token); + throw new Error("A multiscan supervisor is already running."); + } return async () => await releaseLock(path, token); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; @@ -296,34 +307,49 @@ async function acquireLock(output: string): Promise<() => Promise> { const existing = await readLockOwner(path); if (existing !== null && processIsRunning(existing.pid)) { - const current = await readLockOwner(path); - if ( - current !== null && - current.pid === existing.pid && - current.token === existing.token - ) { + throw new Error("A multiscan supervisor is already running."); + } + + try { + await link(pending, recovery); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + const recovering = await readLockOwner(recovery); + if (recovering !== null && processIsRunning(recovering.pid)) { throw new Error("A multiscan supervisor is already running."); } + await rm(recovery, { force: true }); continue; } - const stale = join(output, `.lock.stale-${randomUUID()}`); try { - await rename(path, stale); + const current = await readLockOwner(path); + if (current !== null && processIsRunning(current.pid)) { + throw new Error("A multiscan supervisor is already running."); + } + const metadata = await lstat(path).catch((error: unknown) => { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + }); + if (metadata === null) continue; + if (metadata.isDirectory()) { + const stale = join(output, `.lock.stale-${randomUUID()}`); + await rename(path, stale); + try { + await rename(pending, path); + } finally { + await rm(stale, { recursive: true, force: true }); + } + } else { + await rename(pending, path); + } + return async () => await releaseLock(path, token); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; throw error; + } finally { + await releaseLock(recovery, token); } - const moved = await readLockOwner(stale); - if (moved !== null && processIsRunning(moved.pid)) { - try { - await rename(stale, path); - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; - } - throw new Error("A multiscan supervisor is already running."); - } - await rm(stale, { recursive: true, force: true }); } } finally { await rm(pending, { force: true }); @@ -373,6 +399,7 @@ async function readLockOwner(path: string): Promise { !("pid" in value) || !Number.isSafeInteger(value.pid) || (value.pid as number) <= 0 || + (value.pid as number) > MAX_LOCK_OWNER_PID || ("token" in value && typeof value.token !== "string") ) { return null; @@ -397,8 +424,23 @@ function processIsRunning(pid: number): boolean { async function releaseLock(path: string, token: string): Promise { const owner = await readLockOwner(path); - if (owner?.token === token) { - await rm(path, { force: true }); + if (owner?.token !== token) return; + + const released = `${path}.released-${randomUUID()}`; + try { + await rename(path, released); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return; + throw error; + } + + try { + const moved = await readLockOwner(released); + if (moved?.token !== token) { + await link(released, path); + } + } finally { + await rm(released, { force: true }); } } diff --git a/sdk/typescript/tests-ts/multiscan.test.ts b/sdk/typescript/tests-ts/multiscan.test.ts index d08878cf..91431ae1 100644 --- a/sdk/typescript/tests-ts/multiscan.test.ts +++ b/sdk/typescript/tests-ts/multiscan.test.ts @@ -1,10 +1,12 @@ import { execFileSync } from "node:child_process"; +import { renameSync, writeFileSync } from "node:fs"; import { access, appendFile, lstat, mkdir, mkdtemp, + open, readFile, readdir, rm, @@ -466,7 +468,17 @@ describe("multiscan", () => { return await completedScan(scanOptions.outputDir!); }); - for (const owner of [undefined, '{"pid":', '{"pid":"invalid"}']) { + for (const owner of [ + undefined, + '{"pid":', + '{"pid":"invalid"}', + '{"pid":0}', + '{"pid":-1}', + '{"pid":2147483648}', + '{"pid":9007199254740991}', + '{"pid":1,"token":false}', + `{"pid":1,"padding":"${"x".repeat(4096)}"}`, + ]) { await mkdir(lock); if (owner !== undefined) { await writeFile(join(lock, "owner.json"), owner); @@ -482,11 +494,258 @@ describe("multiscan", () => { expect( (await readdir(paths.output)).some( (name) => - name.startsWith(".lock.pending-") || name.startsWith(".lock.stale-"), + name.startsWith(".lock.pending-") || + name.startsWith(".lock.stale-") || + name.startsWith(".lock.recovery") || + name.startsWith(".lock.released-"), ), ).toBe(false); }); + test("serializes supervisors while recovering stale file and legacy locks", async () => { + for (const legacy of [false, true]) { + const paths = await fixture(); + const source = await repository( + paths.root, + legacy ? "legacy-contention" : "file-contention", + ); + await writeFile( + paths.input, + `id,repository,revision\ntarget,${source.path},${source.revision}\n`, + ); + await mkdir(paths.output); + const lock = join(paths.output, ".lock"); + const owner = JSON.stringify({ + pid: 999_999_999, + ...(legacy ? {} : { token: "crashed-supervisor" }), + }); + if (legacy) { + await mkdir(lock); + await writeFile(join(lock, "owner.json"), owner); + } else { + await writeFile(lock, owner); + } + + let active = 0; + let maximum = 0; + let rejected = 0; + let signalStarted!: () => void; + let signalBlocked!: () => void; + let release!: () => void; + const started = new Promise((resolve) => { + signalStarted = resolve; + }); + const blocked = new Promise((resolve) => { + signalBlocked = resolve; + }); + const completion = new Promise((resolve) => { + release = resolve; + }); + const security = client(async (_repository, scanOptions = {}) => { + await completion; + return await completedScan(scanOptions.outputDir!); + }); + const attempts = Array.from({ length: 24 }, () => + runMultiscan( + options(paths, security, { + createSecurity: () => { + active += 1; + maximum = Math.max(maximum, active); + signalStarted(); + return client(security.run, async () => { + active -= 1; + }); + }, + }), + ).catch((error: unknown) => { + rejected += 1; + if (rejected === 23) signalBlocked(); + throw error; + }), + ); + const settledAttempts = Promise.allSettled(attempts); + + await started; + let timeout: ReturnType | undefined; + try { + await Promise.race([ + blocked, + new Promise((_resolve, reject) => { + timeout = setTimeout(() => { + reject(new Error("Competing supervisors were not rejected.")); + }, 10_000); + }), + ]); + } finally { + clearTimeout(timeout); + release(); + } + const settled = await settledAttempts; + + expect(maximum).toBe(1); + expect( + settled.filter((attempt) => attempt.status === "fulfilled"), + ).toHaveLength(1); + await expect(access(lock)).rejects.toThrow(); + expect( + (await readdir(paths.output)).some((name) => name.startsWith(".lock")), + ).toBe(false); + } + }); + + test("preserves locks published while another supervisor checks recovery", async () => { + const paths = await fixture(); + const source = await repository(paths.root, "lock-replacement"); + await writeFile( + paths.input, + `id,repository,revision\nlock-replacement,${source.path},${source.revision}\n`, + ); + await mkdir(paths.output); + const lock = join(paths.output, ".lock"); + await writeFile(lock, JSON.stringify({ pid: process.pid, token: "moved" })); + const originalKill = process.kill; + let checks = 0; + process.kill = ((pid: number, signal?: NodeJS.Signals | number) => { + if (pid !== process.pid) return originalKill.call(process, pid, signal); + checks += 1; + if (checks === 1) { + const error = new Error( + "simulated stale owner", + ) as NodeJS.ErrnoException; + error.code = "ESRCH"; + throw error; + } + if (checks === 2) { + writeFileSync(lock, JSON.stringify({ pid, token: "replacement" })); + } + return true; + }) as typeof process.kill; + + try { + await expect( + runMultiscan( + options( + paths, + client(async (_repository, scanOptions = {}) => + completedScan(scanOptions.outputDir!), + ), + ), + ), + ).rejects.toThrow("A multiscan supervisor is already running."); + expect(JSON.parse(await readFile(lock, "utf8"))).toMatchObject({ + pid: process.pid, + token: "replacement", + }); + } finally { + process.kill = originalKill; + } + }); + + test("preserves a replacement lock substituted during ownership release", async () => { + const paths = await fixture(); + const source = await repository(paths.root, "lock-release"); + await writeFile( + paths.input, + `id,repository,revision\nlock-release,${source.path},${source.revision}\n`, + ); + let signalStarted!: () => void; + let release!: () => void; + const started = new Promise((resolve) => { + signalStarted = resolve; + }); + const completion = new Promise((resolve) => { + release = resolve; + }); + const security = client(async (_repository, scanOptions = {}) => { + signalStarted(); + await completion; + return await completedScan(scanOptions.outputDir!); + }); + const running = runMultiscan(options(paths, security)); + await started; + + const lock = join(paths.output, ".lock"); + const current = JSON.parse(await readFile(lock, "utf8")) as { + token: string; + }; + const originalParse = JSON.parse; + let replaced = false; + JSON.parse = ((...arguments_: Parameters): unknown => { + const value = originalParse(...arguments_) as unknown; + if ( + !replaced && + typeof value === "object" && + value !== null && + "token" in value && + value.token === current.token + ) { + replaced = true; + const replacement = join(paths.output, ".lock-replacement"); + writeFileSync( + replacement, + JSON.stringify({ pid: process.pid, token: "new-supervisor" }), + ); + renameSync(replacement, lock); + } + return value; + }) as typeof JSON.parse; + + try { + release(); + await expect(running).resolves.toMatchObject({ completed: 1, failed: 0 }); + expect(replaced).toBe(true); + expect(originalParse(await readFile(lock, "utf8"))).toMatchObject({ + pid: process.pid, + token: "new-supervisor", + }); + } finally { + JSON.parse = originalParse; + } + }); + + test("cleans pending ownership files when synchronization fails", async () => { + const paths = await fixture(); + const source = await repository(paths.root, "lock-sync-failure"); + await writeFile( + paths.input, + `id,repository,revision\nlock-sync-failure,${source.path},${source.revision}\n`, + ); + await mkdir(paths.output); + const probePath = join(paths.output, ".probe"); + const probe = await open(probePath, "wx"); + const prototype = Object.getPrototypeOf(probe) as { + sync(): Promise; + }; + await probe.close(); + await rm(probePath); + const originalSync = prototype.sync; + prototype.sync = async () => { + const error = new Error( + "simulated lock sync failure", + ) as NodeJS.ErrnoException; + error.code = "EIO"; + throw error; + }; + + try { + await expect( + runMultiscan( + options( + paths, + client(async (_repository, scanOptions = {}) => + completedScan(scanOptions.outputDir!), + ), + ), + ), + ).rejects.toThrow("simulated lock sync failure"); + expect( + (await readdir(paths.output)).some((name) => name.startsWith(".lock")), + ).toBe(false); + } finally { + prototype.sync = originalSync; + } + }); + test("retries a failed attempt and records both durable receipts", async () => { const paths = await fixture(); const source = await repository(paths.root, "retry");