diff --git a/server/src/__tests__/workspace-worktree-concurrency.test.ts b/server/src/__tests__/workspace-worktree-concurrency.test.ts new file mode 100644 index 00000000000..5b212633fbb --- /dev/null +++ b/server/src/__tests__/workspace-worktree-concurrency.test.ts @@ -0,0 +1,143 @@ +import { execFile } from "node:child_process"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import { afterEach, describe, expect, it } from "vitest"; +import { realizeExecutionWorkspace, worktreeCreateLimits } from "../services/workspace-runtime.js"; + +const execFileAsync = promisify(execFile); +const git = async (cwd: string, args: string[]) => (await execFileAsync("git", args, { cwd })).stdout.trim(); + +describe("concurrent git worktree creation", () => { + const roots: string[] = []; + afterEach(async () => { + for (const root of roots.splice(0)) await fs.rm(root, { recursive: true, force: true }); + }); + + /** A clone of an origin repository: each new branch from origin/main records its upstream in .git/config. */ + async function clonedRepository() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-worktree-race-")); + roots.push(root); + const origin = path.join(root, "origin"); + await fs.mkdir(origin); + await git(origin, ["init", "--initial-branch", "main"]); + await git(origin, ["config", "user.email", "paperclip@example.com"]); + await git(origin, ["config", "user.name", "Paperclip Test"]); + await fs.writeFile(path.join(origin, "README.md"), "hello\n"); + await git(origin, ["add", "README.md"]); + await git(origin, ["commit", "-m", "Initial commit"]); + await git(root, ["clone", "--quiet", origin, "repo"]); + return path.join(root, "repo"); + } + + it("creates every worktree when several tasks start on one repository at once", async () => { + // Several rounds of simultaneous creates make the shared .git/config race certain without serialization. + for (let round = 0; round < 3; round += 1) await createConcurrently(await clonedRepository(), round); + }, 120_000); + + it("waits out another process's brief config lock instead of failing the create", async () => { + const repoRoot = await clonedRepository(); + const lock = path.join(repoRoot, ".git", "config.lock"); + await fs.writeFile(lock, ""); + const release = setTimeout(() => { void fs.rm(lock, { force: true }); }, 300); + try { + const workspace = await realizeExecutionWorkspace({ + base: { baseCwd: repoRoot, source: "project_primary", projectId: "project-1", workspaceId: "workspace-1", + repoUrl: null, repoRef: "origin/main" }, + config: { workspaceStrategy: { type: "git_worktree", branchTemplate: "{{issue.identifier}}-{{slug}}" } }, + issue: { id: "issue-lock", identifier: "PAP-900", title: "Locked config" }, + agent: { id: "agent-1", name: "Codex Coder", companyId: "company-1" }, + }); + expect(workspace).toMatchObject({ created: true, branchCreatedByRuntime: true }); + expect(await git(repoRoot, ["config", "--get", `branch.${workspace.branchName}.remote`])).toBe("origin"); + } finally { + clearTimeout(release); + } + }, 60_000); + + const realize = (repoRoot: string, identifier: string) => realizeExecutionWorkspace({ + base: { baseCwd: repoRoot, source: "project_primary", projectId: "project-1", workspaceId: "workspace-1", + repoUrl: null, repoRef: "origin/main" }, + config: { workspaceStrategy: { type: "git_worktree", branchTemplate: "{{issue.identifier}}-{{slug}}" } }, + issue: { id: `issue-${identifier}`, identifier, title: `Task ${identifier}` }, + agent: { id: "agent-1", name: "Codex Coder", companyId: "company-1" }, + }); + + it("leaves no half-created branch when the lock outlasts every retry, so a later retry creates it cleanly", async () => { + const repoRoot = await clonedRepository(); + const lock = path.join(repoRoot, ".git", "config.lock"); + await fs.writeFile(lock, ""); + await expect(realize(repoRoot, "PAP-910")).rejects.toThrow(/could not lock config file/); + await fs.rm(lock, { force: true }); + const workspace = await realize(repoRoot, "PAP-910"); + expect(workspace).toMatchObject({ created: true, branchCreatedByRuntime: true }); + expect(await git(repoRoot, ["config", "--get", `branch.${workspace.branchName}.remote`])).toBe("origin"); + }, 60_000); + + it("stops a stuck create at its deadline so later creates on the repository proceed", async () => { + const repoRoot = await clonedRepository(); + const hooks = path.join(path.dirname(repoRoot), "hooks"); + await fs.mkdir(hooks); + await fs.writeFile(path.join(hooks, "post-checkout"), "#!/bin/sh\nsleep 30\n", { mode: 0o755 }); + await git(repoRoot, ["config", "core.hooksPath", hooks]); + const original = worktreeCreateLimits.timeoutMs; + worktreeCreateLimits.timeoutMs = 1_000; + try { + const stuck = realize(repoRoot, "PAP-920"); + const startedAt = Date.now(); + await expect(stuck).rejects.toThrow(/timed out/); + await git(repoRoot, ["config", "--unset", "core.hooksPath"]); + const next = await realize(repoRoot, "PAP-921"); + expect(next).toMatchObject({ created: true }); + expect(Date.now() - startedAt).toBeLessThan(20_000); + } finally { + worktreeCreateLimits.timeoutMs = original; + } + }, 60_000); + + it("stops a checkout helper that ignores SIGTERM before the next create runs", async () => { + const repoRoot = await clonedRepository(); + const hooks = path.join(path.dirname(repoRoot), "hooks"); + const pidFile = path.join(path.dirname(repoRoot), "hook.pid"); + await fs.mkdir(hooks); + await fs.writeFile(path.join(hooks, "post-checkout"), `#!/bin/sh\ntrap '' TERM\necho $$ > '${pidFile}'\nsleep 30\n`, { mode: 0o755 }); + await git(repoRoot, ["config", "core.hooksPath", hooks]); + const original = worktreeCreateLimits.timeoutMs; + worktreeCreateLimits.timeoutMs = 1_000; + try { + await expect(realize(repoRoot, "PAP-930")).rejects.toThrow(/timed out/); + const pid = Number((await fs.readFile(pidFile, "utf8")).trim()); + // Gone or a zombie (exited, not yet reaped) both mean it no longer runs. + const running = async () => { + const state = await execFileAsync("ps", ["-o", "stat=", "-p", String(pid)]).then(result => result.stdout.trim(), () => ""); + return state !== "" && !state.startsWith("Z"); + }; + const deadline = Date.now() + 5_000; + while (await running() && Date.now() < deadline) await new Promise(resolve => setTimeout(resolve, 100)); + expect(await running()).toBe(false); + } finally { + worktreeCreateLimits.timeoutMs = original; + } + }, 60_000); + + async function createConcurrently(repoRoot: string, round: number) { + const tasks = Array.from({ length: 10 }, (_, index) => round * 10 + index + 1); + const workspaces = await Promise.all(tasks.map(index => realizeExecutionWorkspace({ + base: { baseCwd: repoRoot, source: "project_primary", projectId: "project-1", workspaceId: "workspace-1", + repoUrl: null, repoRef: "origin/main" }, + config: { workspaceStrategy: { type: "git_worktree", branchTemplate: "{{issue.identifier}}-{{slug}}" } }, + issue: { id: `issue-${index}`, identifier: `PAP-${index}`, title: `Concurrent task ${index}` }, + agent: { id: "agent-1", name: "Codex Coder", companyId: "company-1" }, + }))); + for (const [position, workspace] of workspaces.entries()) { + // Each task got a fresh branch it owns, not a fallback attach after a failed create. + expect(workspace).toMatchObject({ created: true, branchCreatedByRuntime: true }); + expect(await git(workspace.cwd, ["rev-parse", "--abbrev-ref", "HEAD"])).toBe(workspace.branchName); + expect(workspace.branchName).toContain(`PAP-${tasks[position]}`); + // The degraded state this guards against: a branch without its upstream. + expect(await git(repoRoot, ["config", "--get", `branch.${workspace.branchName}.remote`])).toBe("origin"); + } + expect(new Set(workspaces.map(workspace => workspace.cwd)).size).toBe(tasks.length); + } +}); diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 9f6d949d54e..fc7196c75b7 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -887,6 +887,8 @@ async function executeProcess(input: { env?: NodeJS.ProcessEnv; maxStdoutBytes?: number; maxStderrBytes?: number; + /** Stops the child (SIGTERM, then SIGKILL 5 s later) and reports a failure after this long. */ + timeoutMs?: number; }): Promise<{ stdout: string; stderr: string; @@ -901,10 +903,13 @@ async function executeProcess(input: { stderr: ProcessOutputAccumulator; code: number | null; }>((resolve, reject) => { + // A deadline-bound command runs in its own process group, so its hooks and + // helpers stop with it instead of holding the output pipes open. const child = spawn(input.command, input.args, { cwd: input.cwd, stdio: ["ignore", "pipe", "pipe"], env: input.env ?? process.env, + detached: input.timeoutMs !== undefined && process.platform !== "win32", }); const stdout = createProcessOutputCapture(input.maxStdoutBytes ?? DEFAULT_EXECUTE_PROCESS_OUTPUT_BYTES); const stderr = createProcessOutputCapture(input.maxStderrBytes ?? DEFAULT_EXECUTE_PROCESS_OUTPUT_BYTES); @@ -914,8 +919,42 @@ async function executeProcess(input: { child.stderr?.on("data", (chunk) => { stderr.append(String(chunk)); }); + let timedOut = false; + let killTimer: ReturnType | null = null; + const signal = (name: NodeJS.Signals) => { + try { + if (child.pid && process.platform !== "win32") process.kill(-child.pid, name); + else child.kill(name); + } catch { /* already exited */ } + }; + let settled = false; + const finish = (code: number | null) => { + if (settled) return; + settled = true; + if (deadline) clearTimeout(deadline); + if (killTimer) clearTimeout(killTimer); + if (timedOut) stderr.append(`\n${input.command} timed out after ${input.timeoutMs} ms`); + resolve({ stdout, stderr, code: timedOut ? null : code }); + }; + const deadline = input.timeoutMs === undefined ? null : setTimeout(() => { + timedOut = true; + signal("SIGTERM"); + killTimer = setTimeout(() => signal("SIGKILL"), 5_000); + killTimer.unref?.(); + }, input.timeoutMs); + deadline?.unref?.(); child.on("error", reject); - child.on("close", (code) => resolve({ stdout, stderr, code })); + child.on("close", finish); + // After a deadline the group is signalled; do not wait for pipes a straggler may hold. + child.on("exit", (code) => { + if (!timedOut) return; + // Git can exit while a helper that ignores SIGTERM keeps running; stop + // the whole group now so nothing still writes when the queue moves on. + signal("SIGKILL"); + child.stdout?.destroy(); + child.stderr?.destroy(); + finish(code); + }); }); const stdout = proc.stdout.finish(); const stderr = proc.stderr.finish(); @@ -3112,6 +3151,68 @@ async function runWorkspaceCommand(input: { ); } +/** + * `git worktree add` writes the repository's shared `.git/config` (the new + * branch's upstream). Two creates on one repository at once fail with "could + * not lock config file". Creates in this process run one at a time per + * repository, keyed by its git common dir; a create that still meets another + * process's lock retries a few times after a short wait. + */ +const repositoryWorktreeLocks = new Map>(); +export const WORKTREE_CONFIG_LOCK_RETRY_DELAYS_MS = [100, 250, 500, 1_000, 2_000] as const; +/** A stuck create must not hold its repository's queue forever. Tests shorten it. */ +export const worktreeCreateLimits = { timeoutMs: 10 * 60_000 }; + +async function withRepositoryWorktreeLock(cwd: string, work: () => Promise): Promise { + const commonDir = await executeProcess({ command: "git", args: ["rev-parse", "--path-format=absolute", "--git-common-dir"], cwd }) + .then(result => result.code === 0 && result.stdout.trim() ? result.stdout.trim() : path.resolve(cwd)) + .catch(() => path.resolve(cwd)); + const key = await fs.realpath(commonDir).catch(() => commonDir); + const previous = repositoryWorktreeLocks.get(key) ?? Promise.resolve(); + const current = previous.catch(() => undefined).then(work); + const tail = current.catch(() => undefined); + repositoryWorktreeLocks.set(key, tail); + try { + return await current; + } finally { + if (repositoryWorktreeLocks.get(key) === tail) repositoryWorktreeLocks.delete(key); + } +} + +function isGitConfigLockFailure(result: { code: number | null; stderr: string }) { + return result.code !== 0 && /could not lock config file/i.test(result.stderr); +} + +async function executeGitCommand(args: string[], cwd: string) { + if (args[0] !== "worktree" || args[1] !== "add") return executeProcess({ command: "git", args, cwd }); + return withRepositoryWorktreeLock(cwd, async () => { + const branchIndex = args.indexOf("-b"); + const branch = branchIndex > 0 ? args[branchIndex + 1] : undefined; + const startPoint = branch ? args[branchIndex + 3] : undefined; + const baseCommit = branch && startPoint ? await executeProcess({ command: "git", + args: ["rev-parse", "--verify", "--quiet", `${startPoint}^{commit}`], cwd }).then(result => result.code === 0 ? result.stdout.trim() : null) + .catch(() => null) : null; + for (let attempt = 0; ; attempt += 1) { + const result = await executeProcess({ command: "git", args, cwd, timeoutMs: worktreeCreateLimits.timeoutMs }); + if (!isGitConfigLockFailure(result)) return result; + // `worktree add -b` creates the branch, then fails writing its upstream, + // and adds no worktree. The branch did not exist before (otherwise git + // refuses with "already exists" first), so this command created it. + // Remove it, also after the last attempt, so a retry creates it again + // with its upstream. A branch that moved off the base commit belongs to + // someone else meanwhile (a same-issue retry elsewhere) and is kept. + if (branch && baseCommit) { + const tip = await executeProcess({ command: "git", args: ["rev-parse", "--verify", "--quiet", `refs/heads/${branch}`], cwd }) + .then(found => found.code === 0 ? found.stdout.trim() : null).catch(() => null); + if (tip === baseCommit) await executeProcess({ command: "git", args: ["branch", "-D", branch], cwd }).catch(() => undefined); + } + const delayMs = WORKTREE_CONFIG_LOCK_RETRY_DELAYS_MS[attempt]; + if (delayMs === undefined) return result; + await new Promise(resolve => setTimeout(resolve, delayMs)); + } + }); +} + async function recordGitOperation( recorder: WorkspaceOperationRecorder | null | undefined, input: { @@ -3124,7 +3225,9 @@ async function recordGitOperation( }, ): Promise { if (!recorder) { - return runGit(input.args, input.cwd); + const result = await executeGitCommand(input.args, input.cwd); + if (result.code !== 0) throw new Error(result.stderr.trim() || result.stdout.trim() || `git ${input.args.join(" ")} failed`); + return result.stdout.trim(); } let stdout = ""; @@ -3136,11 +3239,7 @@ async function recordGitOperation( cwd: input.cwd, metadata: input.metadata ?? null, run: async () => { - const result = await executeProcess({ - command: "git", - args: input.args, - cwd: input.cwd, - }); + const result = await executeGitCommand(input.args, input.cwd); stdout = result.stdout; stderr = result.stderr; code = result.code;