diff --git a/packages/loopover-miner/lib/repo-clone.js b/packages/loopover-miner/lib/repo-clone.js index b9fff4b0a6..9dd60197d8 100644 --- a/packages/loopover-miner/lib/repo-clone.js +++ b/packages/loopover-miner/lib/repo-clone.js @@ -72,6 +72,57 @@ async function defaultRunGit(args, cwd, timeoutMs) { } } +// Per-repoPath in-process serialization for ensureRepoCloned (#6762). Two attempts for the SAME repo share +// one deterministic base-clone path and mutate it in place (git fetch/checkout/reset --hard); worktree- +// allocator.js only caps the TOTAL active-slot count, never per-repo exclusivity, so without this two +// same-repo attempts can interleave git subprocesses on the same .git dir and corrupt the index/HEAD/refs or +// trip .git/index.lock. `repoCloneLocks` maps a resolved repoPath to the tail of its in-flight promise chain: +// same-repo calls run strictly one after another, while different repoPaths stay fully parallel. The tail +// promise's handlers swallow, so it never rejects -- one failing attempt can neither reject a waiter nor +// wedge the queue -- and the finally drops the entry once the chain drains, keeping the Map bounded. +const repoCloneLocks = new Map(); + +/** + * @template T + * @param {string} repoPath key: the resolved base-clone path the git mutations run against. + * @param {() => Promise} fn the critical section (a single ensureRepoClonedUnlocked run). + * @returns {Promise} + */ +async function withRepoCloneLock(repoPath, fn) { + const previous = repoCloneLocks.get(repoPath) ?? Promise.resolve(); + const run = previous.then(() => fn()); + const tail = run.then( + () => {}, + () => {}, + ); + repoCloneLocks.set(repoPath, tail); + try { + return await run; + } finally { + if (repoCloneLocks.get(repoPath) === tail) repoCloneLocks.delete(repoPath); + } +} + +/** + * Serialize the git mutations of {@link ensureRepoClonedUnlocked} per resolved repo path so concurrent + * same-repo attempts never race the shared base clone (#6762), while different repos still run in parallel. + * Resolves the same `repoPath` the unlocked step computes and uses it as the mutex key; throws (before + * locking) on a malformed `repoFullName`, matching the prior behaviour. + * + * @param {string} repoFullName + * @param {{ + * baseBranch?: string, cloneBaseDir?: string, env?: Record, timeoutMs?: number, + * remoteUrl?: string, runGit?: (args: string[], cwd: string, timeoutMs: number) => Promise<{ ok: boolean, stdout: string, stderr: string }>, + * }} [options] + * @returns {Promise<{ ok: boolean, repoPath: string, error?: string }>} + */ +export async function ensureRepoCloned(repoFullName, options = {}) { + const target = normalizeRepoFullName(repoFullName); + const cloneBaseDir = typeof options.cloneBaseDir === "string" && options.cloneBaseDir.trim() ? options.cloneBaseDir.trim() : resolveRepoCloneBaseDir(options.env); + const repoPath = join(cloneBaseDir, target.owner, target.repo); + return withRepoCloneLock(repoPath, () => ensureRepoClonedUnlocked(repoFullName, options)); +} + /** * Ensure a real, current local clone of `repoFullName` exists at the deterministic per-repo cache path. * First use: `git clone`. Subsequent use: `git fetch origin` + hard-reset the base branch to @@ -84,7 +135,7 @@ async function defaultRunGit(args, cwd, timeoutMs) { * }} [options] * @returns {Promise<{ ok: boolean, repoPath: string, error?: string }>} */ -export async function ensureRepoCloned(repoFullName, options = {}) { +async function ensureRepoClonedUnlocked(repoFullName, options = {}) { const target = normalizeRepoFullName(repoFullName); const baseBranch = typeof options.baseBranch === "string" && options.baseBranch.trim() ? options.baseBranch.trim() : DEFAULT_BASE_BRANCH; const cloneBaseDir = typeof options.cloneBaseDir === "string" && options.cloneBaseDir.trim() ? options.cloneBaseDir.trim() : resolveRepoCloneBaseDir(options.env); diff --git a/test/unit/miner-repo-clone.test.ts b/test/unit/miner-repo-clone.test.ts index 847541eaf2..9ede31c80a 100644 --- a/test/unit/miner-repo-clone.test.ts +++ b/test/unit/miner-repo-clone.test.ts @@ -1,5 +1,5 @@ import { execFileSync } from "node:child_process"; -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; @@ -195,3 +195,116 @@ describe("ensureRepoCloned (#5132)", () => { expect(runGitCalls).toBe(0); }); }); + +describe("ensureRepoCloned per-repo concurrency guard (#6762)", () => { + // Drains the microtask queue: a setImmediate callback only fires once no microtasks remain ready, so + // awaiting this lets every already-schedulable git op run while leaving anything still blocked on a gate + // (or queued behind the mutex) untouched. + const flush = () => new Promise((resolve) => setImmediate(resolve)); + + it("REGRESSION: serializes two concurrent ensureRepoCloned calls for the SAME repo (no interleaved git ops)", async () => { + // Both concurrent calls share one injected runGit whose first invocation blocks on `firstGate`. WITHOUT + // the per-repo mutex the second call enters its own git op immediately and `events` shows two + // "start:clone" before either ends; WITH the guard the second cannot start any git op until the first + // fully settles, so exactly one op is ever in flight. + const root = tempRoot("loopover-miner-repo-clone-concurrent-same-"); + const cloneBaseDir = join(root, "cache"); + const events: string[] = []; + let releaseFirst!: () => void; + const firstGate = new Promise((resolve) => { + releaseFirst = resolve; + }); + let firstBlocked = false; + const runGit = async (args: string[]) => { + events.push(`start:${args[0]}`); + if (!firstBlocked) { + firstBlocked = true; + await firstGate; + } + events.push(`end:${args[0]}`); + return { ok: true, stdout: "", stderr: "" }; + }; + + const first = ensureRepoCloned("acme/widgets", { cloneBaseDir, remoteUrl: "unused", runGit }); + const second = ensureRepoCloned("acme/widgets", { cloneBaseDir, remoteUrl: "unused", runGit }); + + // Only the first call may have reached git; the second must be queued behind the per-repo lock. + await flush(); + expect(events).toEqual(["start:clone"]); + + releaseFirst(); + const [firstResult, secondResult] = await Promise.all([first, second]); + expect(firstResult.ok).toBe(true); + expect(secondResult.ok).toBe(true); + // Strict, non-overlapping ordering: first runs start->end fully before second starts. + expect(events).toEqual(["start:clone", "end:clone", "start:clone", "end:clone"]); + }); + + it("does NOT serialize across DIFFERENT repos -- they run in parallel", async () => { + // repo-a's git op blocks; repo-b's must still proceed (different repoPath => different lock), proving the + // guard is per-repo rather than a single global mutex. + const root = tempRoot("loopover-miner-repo-clone-concurrent-diff-"); + const cloneBaseDir = join(root, "cache"); + const started: string[] = []; + let releaseA!: () => void; + const gateA = new Promise((resolve) => { + releaseA = resolve; + }); + const runGitFor = (name: string) => async () => { + started.push(name); + if (name === "a") await gateA; + return { ok: true, stdout: "", stderr: "" }; + }; + + const a = ensureRepoCloned("acme/repo-a", { cloneBaseDir, remoteUrl: "unused", runGit: runGitFor("a") }); + const b = ensureRepoCloned("acme/repo-b", { cloneBaseDir, remoteUrl: "unused", runGit: runGitFor("b") }); + + await flush(); + // repo-b advanced into git even though repo-a is still blocked -> not serialized against each other. + expect(started).toContain("b"); + + releaseA(); + const [aResult, bResult] = await Promise.all([a, b]); + expect(aResult.ok).toBe(true); + expect(bResult.ok).toBe(true); + }); + + it("releases the per-repo lock when a call throws, so a later same-repo call still proceeds", async () => { + const root = tempRoot("loopover-miner-repo-clone-concurrent-throw-"); + const cloneBaseDir = join(root, "cache"); + const throwing = async () => { + throw new Error("git exploded"); + }; + await expect(ensureRepoCloned("acme/widgets", { cloneBaseDir, remoteUrl: "unused", runGit: throwing })).rejects.toThrow("git exploded"); + + // If the lock were not released on throw, this second call would block forever (test would time out). + const ok = async () => ({ ok: true, stdout: "", stderr: "" }); + const result = await ensureRepoCloned("acme/widgets", { cloneBaseDir, remoteUrl: "unused", runGit: ok }); + expect(result.ok).toBe(true); + }); + + it("keys the lock off the env-resolved base dir when no cloneBaseDir option is given", async () => { + // Exercises the wrapper's env-fallback path for the lock key (no explicit cloneBaseDir option). + const root = tempRoot("loopover-miner-repo-clone-concurrent-envdir-"); + const ok = async () => ({ ok: true, stdout: "", stderr: "" }); + const result = await ensureRepoCloned("acme/widgets", { env: { LOOPOVER_MINER_REPO_CLONE_DIR: root }, remoteUrl: "unused", runGit: ok }); + expect(result.ok).toBe(true); + expect(result.repoPath).toBe(join(root, "acme", "widgets")); + }); + + it("propagates real git stderr and falls back to a default across the fetch/checkout/reset steps", async () => { + // existsSync(repoPath) true => fetch/checkout/reset path, driven entirely with injected runGit (fast, + // deterministic). Covers both the real-stderr and empty-stderr fallback branch of each step. + const root = tempRoot("loopover-miner-repo-clone-concurrent-stderr-"); + const cloneBaseDir = join(root, "cache"); + const repoPath = join(cloneBaseDir, "acme", "widgets"); + mkdirSync(repoPath, { recursive: true }); + + const failOn = (step: string, stderr: string) => async (args: string[]) => (args[0] === step ? { ok: false, stdout: "", stderr } : { ok: true, stdout: "", stderr: "" }); + + expect((await ensureRepoCloned("acme/widgets", { cloneBaseDir, runGit: failOn("fetch", "") })).error).toBe("git_fetch_failed"); + expect((await ensureRepoCloned("acme/widgets", { cloneBaseDir, runGit: failOn("fetch", "boom-fetch") })).error).toBe("boom-fetch"); + expect((await ensureRepoCloned("acme/widgets", { cloneBaseDir, runGit: failOn("checkout", "boom-checkout") })).error).toBe("boom-checkout"); + expect((await ensureRepoCloned("acme/widgets", { cloneBaseDir, runGit: failOn("reset", "boom-reset") })).error).toBe("boom-reset"); + }); +});