fix(miner): serialize per-repo base clone to prevent concurrent git races (#6917)

ensureRepoCloned resolves one deterministic path per repoFullName and
mutates it in place (git fetch/checkout/reset --hard) with no locking,
while worktree-allocator only caps the total active-slot count and never
enforces per-repo exclusivity. Two attempts for the same repo could
therefore run these git commands against the shared base clone
concurrently, corrupting the index/HEAD/refs or tripping .git/index.lock.

Add a per-repoPath in-process async mutex: a Map of repoPath to the tail
of an in-flight promise chain serializes same-repo ensureRepoCloned calls
while different repoPaths still run in parallel. The lock is released on
both success and failure (finally), so one failing attempt can neither
reject a waiter nor wedge the queue, and the entry is dropped once the
chain drains.

Closes #6762

Co-authored-by: e11734937-beep <e11734937-beep@users.noreply.github.com>

Author: e11734937-beep

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<T>} fn the critical section (a single ensureRepoClonedUnlocked run).
+ * @returns {Promise<T>}
+ */
+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<string, string | undefined>, 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);

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<void>((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<void>((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<void>((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");
+  });
+});