From b6bd69fe633297e335858321bddbaa510dd46fed Mon Sep 17 00:00:00 2001 From: Jake Leventhal Date: Thu, 8 Oct 2026 08:53:50 -0400 Subject: [PATCH] fix(server): bulk worktree launches no longer fail on Git races Launching many threads into new worktrees at once failed most of them: concurrent fetches of the base lost the ref lock, branch and path names collided, and concurrent writes to the shared Git config failed on its lock. Retry hit the same collisions again. Concurrent fetches of one ref now share a single running fetch. A taken branch or path gets the next `-1`, `-2` suffix, the same style renames use, while Git stays the arbiter. Config writes and branch renames take a per-repository lock, so checkouts stay concurrent and concurrent renames pick distinct names. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../ThreadLaunchService.test.ts | 368 ++++++++++++++++-- .../orchestration-v2/ThreadLaunchService.ts | 43 +- apps/server/src/vcs/GitVcsDriverCore.test.ts | 102 +++++ apps/server/src/vcs/GitVcsDriverCore.ts | 83 +++- 4 files changed, 538 insertions(+), 58 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index dfa02f4f514b..cc65eb02c260 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -34,12 +34,14 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; import * as TestClock from "effect/testing/TestClock"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ProjectStore from "./ProjectStore.ts"; import * as ProjectService from "../project/ProjectService.ts"; @@ -1235,6 +1237,289 @@ it.effect("keeps an explicit branch name instead of generating one", () => }), ); +it.effect.each([false, true])( + "bulk launches get independent real worktrees with existing branch %s", + (branchExists) => + Effect.gen(function* () { + const driver = yield* GitVcsDriver.GitVcsDriver; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "launch-worktree-collisions-" }); + const worktreesDirectory = yield* fs.makeTempDirectoryScoped({ prefix: "launch-worktrees-" }); + const git = (args: ReadonlyArray) => + driver.execute({ operation: "ThreadLaunchService.test", cwd, args }); + yield* git(["init", "--initial-branch=main"]); + yield* git([ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.com", + "commit", + "--allow-empty", + "-m", + "initial", + ]); + const existingPath = path.join(worktreesDirectory, "existing"); + if (branchExists) { + yield* driver.createWorktree({ + cwd, + refName: "main", + newRefName: "feature/shared", + path: existingPath, + }); + yield* fs.writeFileString(path.join(existingPath, "keep.txt"), "existing uncommitted work"); + } + const harness = makeHarness({ + createWorktree: (input, options) => + driver.createWorktree({ ...input, cwd }, { ...options, worktreesDirectory }), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* Effect.all( + Array.from({ length: 12 }, (_, index) => { + const command = `command:launch:bulk-${index}`; + return launches + .launch( + launchInput({ + command, + thread: `thread:launch:bulk-${index}`, + message: "Independent work", + workspace: { type: "worktree", baseRef: "main", branch: "feature/shared" }, + }), + ) + .pipe( + Effect.flatMap((result) => + threads.streamStoredEventsFrom({ threadId: result.threadId }).pipe( + Stream.filter( + (stored) => + (stored.commandId === CommandId.make(`${command}:release`) || + stored.commandId === CommandId.make(`${command}:fail`)) && + stored.event.type === "run.updated", + ), + Stream.runHead, + Effect.andThen(threads.getThreadProjection(result.threadId)), + ), + ), + ); + }), + { concurrency: "unbounded" }, + ); + assert.equal(new Set(launched.map((projection) => projection.thread.branch)).size, 12); + assert.equal( + new Set(launched.map((projection) => projection.thread.worktreePath)).size, + 12, + ); + for (const projection of launched) { + assert.equal( + projection.runs[0]?.status, + "starting", + projection.turnItems.find((item) => item.type === "error")?.failure.message, + ); + assert.isTrue(yield* fs.exists(projection.thread.worktreePath!)); + const base = yield* git(["config", `branch.${projection.thread.branch}.gh-merge-base`]); + assert.equal(base.stdout.trim(), "main"); + } + assert.deepEqual( + launched.map((projection) => projection.thread.branch).toSorted(), + Array.from({ length: 12 }, (_, index) => + branchExists || index > 0 + ? `feature/shared-${branchExists ? index + 1 : index}` + : "feature/shared", + ).toSorted(), + ); + if (branchExists) { + assert.equal( + yield* fs.readFileString(path.join(existingPath, "keep.txt")), + "existing uncommitted work", + ); + } + assert.equal(harness.removeWorktree.mock.calls.length, 0); + }).pipe(Effect.provide(harness.layer)); + }).pipe( + Effect.provide( + GitVcsDriver.layer.pipe( + Layer.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "launch-collision-test-" }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ), + ), +); + +it.effect.each([ + { branch: "t3/abcd1234", flat: false }, + { branch: "t3-abcd1234", flat: true }, +])( + "keeps temporary naming and background rename after a collision on $branch", + ({ branch, flat }) => { + let attempts = 0; + const harness = makeHarness({ + hasCommit: () => Effect.succeed(flat), + createWorktree: (input) => + ++attempts === 1 + ? Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.createWorktree", + command: "git", + cwd: project.workspaceRoot, + detail: "git worktree add failed", + reason: "branch_already_exists", + exitCode: 128, + }), + ) + : Effect.succeed({ + worktree: { path: "/repo-worktrees/fresh", refName: input.newRefName! }, + }), + }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const command = `command:launch:temp-collision-${flat}`; + const launched = yield* launches.launch( + launchInput({ + command, + thread: `thread:launch:temp-collision-${flat}`, + message: "Name this worktree", + workspace: { type: "worktree", baseRef: "main", branch }, + }), + ); + yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( + Stream.filter((stored) => stored.commandId === CommandId.make(`${command}:branch-rename`)), + Stream.runHead, + ); + assert.equal(harness.createWorktree.mock.calls.length, 2); + const replacement = harness.createWorktree.mock.calls[1]![0].newRefName!; + assert.notEqual(replacement, branch); + assert.match(replacement, flat ? /^t3-[0-9a-f]{8}$/u : /^t3\/[0-9a-f]{8}$/u); + assert.equal(harness.renameBranch.mock.calls[0]![0].oldBranch, replacement); + assert.equal( + (yield* threads.getThreadProjection(launched.threadId)).thread.branch, + "generated-branch", + ); + assert.equal(harness.removeWorktree.mock.calls.length, 0); + }).pipe(Effect.provide(harness.layer)); + }, +); + +it.effect.each(["path_already_exists", "authentication_failed", "branch_already_exists"] as const)( + "handles %s without unbounded attempts", + (reason) => { + let attempts = 0; + const harness = makeHarness({ + createWorktree: (input) => { + attempts += 1; + return reason === "path_already_exists" && attempts > 1 + ? Effect.succeed({ + worktree: { path: "/repo-worktrees/fresh", refName: input.newRefName! }, + }) + : Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.createWorktree", + command: "git", + cwd: project.workspaceRoot, + detail: "git worktree add failed", + reason, + exitCode: 128, + }), + ); + }, + }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const command = `command:launch:collision-limit-${reason}`; + const launched = yield* launches.launch( + launchInput({ + command, + thread: `thread:launch:collision-limit-${reason}`, + message: "Prepare worktree", + workspace: { type: "worktree", baseRef: "main", branch: "my-feature" }, + }), + ); + const succeeds = reason === "path_already_exists"; + yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( + Stream.filter( + (stored) => + stored.commandId === CommandId.make(`${command}:${succeeds ? "release" : "fail"}`) && + stored.event.type === "run.updated", + ), + Stream.runHead, + ); + assert.equal(attempts, succeeds ? 2 : reason === "authentication_failed" ? 1 : 101); + const projection = yield* threads.getThreadProjection(launched.threadId); + assert.equal(projection.runs[0]?.status, succeeds ? "starting" : "failed"); + assert.equal(harness.runSetup.mock.calls.length, succeeds ? 1 : 0); + assert.equal(harness.removeWorktree.mock.calls.length, 0); + }).pipe(Effect.provide(harness.layer)); + }, +); + +it.effect("Retry recovers an existing failed run whose requested branch is taken", () => { + let fails = true; + const harness = makeHarness({ + createWorktree: (input) => + fails || input.newRefName === "my-feature" + ? Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.createWorktree", + command: "git", + cwd: project.workspaceRoot, + detail: "git worktree add failed", + reason: "branch_already_exists", + exitCode: 128, + }), + ) + : Effect.succeed({ + worktree: { path: "/repo-worktrees/recovered", refName: input.newRefName! }, + }), + }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:taken-retry", + thread: "thread:launch:taken-retry", + message: "Retry me", + workspace: { type: "worktree", baseRef: "main", branch: "my-feature" }, + }), + ); + yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( + Stream.filter( + (stored) => + stored.commandId === CommandId.make("command:launch:taken-retry:fail") && + stored.event.type === "run.updated", + ), + Stream.runHead, + ); + const failed = yield* threads.getThreadProjection(launched.threadId); + fails = false; + yield* launches.retryPreparation({ + commandId: CommandId.make("command:launch:taken-retry:retry"), + threadId: launched.threadId, + runId: failed.runs[0]!.id, + }); + yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( + Stream.filter( + (stored) => + stored.commandId === CommandId.make("command:launch:taken-retry:retry:release") && + stored.event.type === "run.updated", + ), + Stream.runHead, + ); + const recovered = yield* threads.getThreadProjection(launched.threadId); + assert.equal(recovered.runs.length, 1); + assert.equal(recovered.runs[0]!.id, failed.runs[0]!.id); + assert.equal(recovered.messages.length, 1); + assert.equal(recovered.runs[0]?.status, "starting"); + assert.equal(recovered.thread.branch, "my-feature-1"); + assert.equal(recovered.thread.worktreePath, "/repo-worktrees/recovered"); + assert.equal(recovered.turnItems.find((item) => item.type === "error")?.status, "cancelled"); + }).pipe(Effect.provide(harness.layer)); +}); + it.effect("keeps the temporary branch when branch generation fails", () => Effect.gen(function* () { const harness = makeHarness({ @@ -1484,39 +1769,56 @@ it.effect("a retry reuses a recorded worktree without undoing its branch rename" }).pipe(Effect.provide(harness.layer)); }); -it.effect("removes a worktree that failed before the thread recorded it", () => { - const harness = makeHarness({ - // A checkout that dies after claiming its directory. - createWorktree: (_input, options) => - (options?.progress?.onWorktreeClaimed?.("/repo-worktrees/partial") ?? Effect.void).pipe( - Effect.andThen(Effect.fail(new Error("checkout failed") as never)), - ), - }); - return Effect.gen(function* () { - const launches = yield* ThreadLaunch.ThreadLaunchService; - const threads = yield* ThreadManagement.ThreadManagementService; - const launched = yield* launches.launch( - launchInput({ - command: "command:launch:partial-worktree", - thread: "thread:launch:partial-worktree", - message: "Partial checkout", - workspace: { type: "worktree", baseRef: "main" }, - }), - ); - yield* waitUntil(() => - threads - .getThreadProjection(launched.threadId) - .pipe(Effect.map((projection) => projection.runs[0]?.status === "failed")), - ); - const projection = yield* threads.getThreadProjection(launched.threadId); - // Unrecorded, so a retry would create a second checkout beside it. - assert.equal(projection.thread.worktreePath, null); - assert.deepEqual( - harness.removeWorktree.mock.calls.map(([input]) => input.path), - ["/repo-worktrees/partial"], - ); - }).pipe(Effect.provide(harness.layer)); -}); +it.effect.each([undefined, "branch_already_exists"] as const)( + "removes a claimed worktree after failure %s without creating another", + (reason) => { + const harness = makeHarness({ + // A checkout that dies after claiming its directory. + createWorktree: (_input, options) => + (options?.progress?.onWorktreeClaimed?.("/repo-worktrees/partial") ?? Effect.void).pipe( + Effect.andThen( + Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.createWorktree.configureBaseRef", + command: "git", + cwd: project.workspaceRoot, + detail: "checkout failed", + ...(reason === undefined ? {} : { reason }), + }), + ), + ), + ), + }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:partial-worktree", + thread: "thread:launch:partial-worktree", + message: "Partial checkout", + workspace: { type: "worktree", baseRef: "main" }, + }), + ); + yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( + Stream.filter( + (stored) => + stored.commandId === CommandId.make("command:launch:partial-worktree:fail") && + stored.event.type === "run.updated", + ), + Stream.runHead, + ); + const projection = yield* threads.getThreadProjection(launched.threadId); + // Unrecorded, so a retry would create a second checkout beside it. + assert.equal(projection.thread.worktreePath, null); + assert.equal(harness.createWorktree.mock.calls.length, 1); + assert.deepEqual( + harness.removeWorktree.mock.calls.map(([input]) => input.path), + ["/repo-worktrees/partial"], + ); + }).pipe(Effect.provide(harness.layer)); + }, +); it.effect.each(["worktree", "setup"] as const)( "%s failure keeps the thread and message visible and emits failure items", diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 10af1aef454d..ae65ee237eb2 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -381,13 +381,14 @@ const make = Effect.gen(function* () { branch = flattenTemporaryWorktreeBranchName(branch); } yield* setupTracker.stageStatus(threadId, "checkout", "running"); - const worktree = yield* git - .createWorktree( + const baseRefName = input.workspaceStrategy.baseRef; + const createWorktree = (newRefName: string) => + git.createWorktree( { cwd: project.workspaceRoot, refName: startRef, - newRefName: branch!, - baseRefName: input.workspaceStrategy.baseRef, + newRefName, + baseRefName, path: null, }, { @@ -400,8 +401,38 @@ const make = Effect.gen(function* () { setupTracker.stage(threadId, "checkout", { percent: progress.percent }), }, }, - ) - .pipe(Effect.mapError(mapError(input, "provision-worktree", threadId))); + ); + const desiredBranch = branch!; + // Git claims the branch atomically. A preflight check alone would race + // with other launches, and retries must also escape leftover branches + // and directories from earlier attempts without touching those checkouts. + // Taken names get the same `-1`, `-2` suffixes renames use. + const worktree = yield* Effect.gen(function* () { + let candidate = desiredBranch; + for (let attempt = 1; ; attempt += 1) { + const result = yield* createWorktree(candidate).pipe( + Effect.catchTags({ + GitCommandError: (error) => + attempt <= 100 && + createdWorktreePath === null && + (error.reason === "branch_already_exists" || + error.reason === "path_already_exists") + ? Effect.succeed(null) + : Effect.fail(error), + }), + ); + if (result !== null) return result; + if (isTemporaryWorktreeBranch(desiredBranch)) { + const uuid = yield* randomUuidV4; + candidate = buildTemporaryWorktreeBranchName(() => uuid.replaceAll("-", "")); + if (!desiredBranch.includes("/")) { + candidate = flattenTemporaryWorktreeBranchName(candidate); + } + } else { + candidate = `${desiredBranch}-${attempt}`; + } + } + }).pipe(Effect.mapError(mapError(input, "provision-worktree", threadId))); worktreePath = worktree.worktree.path; branch = worktree.worktree.refName; createdWorktreePath = worktreePath; diff --git a/apps/server/src/vcs/GitVcsDriverCore.test.ts b/apps/server/src/vcs/GitVcsDriverCore.test.ts index a71226c06aee..5fb89ed72525 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.test.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.test.ts @@ -1263,6 +1263,31 @@ it.layer(layerTest)("GitVcsDriver core integration", (it) => { }), ); + it.effect.each(["heads", "tags"] as const)( + "classifies an atomic ref creation collision under refs/%s", + (namespace) => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + const driver = yield* GitVcsDriver.GitVcsDriver; + yield* initRepoWithCommit(cwd); + const ref = `refs/${namespace}/existing`; + yield* git(cwd, ["update-ref", ref, "HEAD"]); + const error = yield* driver + .execute({ + operation: "GitVcsDriver.test.atomicRefCollision", + cwd, + args: ["update-ref", ref, "HEAD", "0".repeat(40)], + env: { LC_ALL: "C" }, + }) + .pipe(Effect.flip); + if (namespace === "heads") { + assert.equal(error.reason, "branch_already_exists"); + } else { + assert.notProperty(error, "reason"); + } + }), + ); + it.effect("leaves a tag collision unclassified rather than calling it a path", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); @@ -2745,6 +2770,38 @@ it.layer(layerTest)("GitVcsDriver core integration", (it) => { }), ); + it.effect("gives concurrent renames to one name distinct suffixes and keeps their config", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd); + const driver = yield* GitVcsDriver.GitVcsDriver; + const sources = Array.from({ length: 6 }, (_, index) => `t3/source-${index}`); + for (const source of sources) { + yield* driver.createRef({ cwd, refName: source }); + yield* git(cwd, ["config", `branch.${source}.gh-merge-base`, "main"]); + } + + const renamed = yield* Effect.all( + sources.map((oldBranch) => + driver.renameBranch({ cwd, oldBranch, newBranch: "feature/shared" }), + ), + { concurrency: "unbounded" }, + ); + + assert.deepEqual(renamed.map((result) => result.branch).toSorted(), [ + "feature/shared", + "feature/shared-1", + "feature/shared-2", + "feature/shared-3", + "feature/shared-4", + "feature/shared-5", + ]); + for (const { branch } of renamed) { + assert.equal(yield* git(cwd, ["config", `branch.${branch}.gh-merge-base`]), "main"); + } + }), + ); + it.effect("returns the existing refName when rename source and target match", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); @@ -3599,6 +3656,51 @@ it.layer(layerTest)("GitVcsDriver core integration", (it) => { }), ); + it.effect("shares one running fetch of a ref between concurrent callers", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + let fetches = 0; + const spawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + if (!ChildProcess.isStandardCommand(command)) + return yield* Effect.die("unexpected command"); + if (command.args[0] !== "fetch") return yield* delegate.spawn(command); + fetches += 1; + yield* Deferred.succeed(started, undefined); + return ChildProcessSpawner.makeHandle({ + ...makeNonRepositoryHandle(), + exitCode: Deferred.await(release).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), + stderr: Stream.empty, + }); + }), + ); + const driver = yield* makeGitVcsDriverCore().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.provide(layerServerConfig), + ); + const fetch = driver.fetchRemote({ cwd, remoteName: "origin", refName: "main" }); + const first = yield* fetch.pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(started); + const joined = yield* Effect.all( + Array.from({ length: 5 }, () => fetch), + { + concurrency: "unbounded", + }, + ).pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(first); + yield* Fiber.join(joined); + assert.equal(fetches, 1); + + // A finished fetch is not reused. + yield* fetch; + assert.equal(fetches, 2); + }), + ); + it.effect("creates a worktree from the latest fetched remote commit", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index 002311fda2cf..c2353dbc5fc0 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -29,6 +29,7 @@ import { type VcsRef, } from "@t3tools/contracts"; import { dedupeRemoteBranchesWithLocalMatches, normalizeGitRemoteUrl } from "@t3tools/shared/git"; +import * as KeyedLock from "@t3tools/shared/KeyedLock"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { compactTraceAttributes } from "@t3tools/shared/observability"; import { decodeJsonResult } from "@t3tools/shared/schemaJson"; @@ -138,6 +139,12 @@ class StatusRemoteRefreshCacheKey extends Data.Class<{ remoteName: string; }> {} +class FetchRemoteKey extends Data.Class<{ + cwd: string; + remoteName: string; + refName: string | undefined; +}> {} + function statusUpstreamRefreshFailureCooldown(consecutiveFailures: number): Duration.Duration { const exponent = Math.max(0, consecutiveFailures - 1); const cooldownMs = @@ -444,6 +451,9 @@ const GIT_FAILURE_REASON_PATTERNS: ReadonlyArray(); const executeRaw: GitVcsDriver.GitVcsDriver["Service"]["execute"] = Effect.fnUntraced( function* (input) { @@ -3527,11 +3540,15 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* remoteNames.toSorted((left, right) => right.length - left.length), ); const baseBranch = parsedBaseRef?.branchName ?? input.baseRefName; - yield* runGit("GitVcsDriver.createWorktree.configureBaseRef", input.cwd, [ - "config", - `branch.${input.newRefName}.gh-merge-base`, - baseBranch, - ]); + // Locked here rather than around the whole call so checkouts stay concurrent. + yield* configLocks.withLock( + yield* resolveGitCommonDir(input.cwd), + runGit("GitVcsDriver.createWorktree.configureBaseRef", input.cwd, [ + "config", + `branch.${input.newRefName}.gh-merge-base`, + baseBranch, + ]), + ); } return { @@ -3721,6 +3738,26 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }); }, ); + // Concurrent fetches of one ref race for the same ref lock and all but one + // fail, as when many threads launch from one base at once. Callers that + // arrive while a fetch is running share it instead. + const fetchRemoteInFlight = yield* Cache.makeWith( + ({ refName, ...input }: FetchRemoteKey) => + fetchRemote(refName === undefined ? input : { ...input, refName }), + { + capacity: Number.POSITIVE_INFINITY, + timeToLive: () => Duration.zero, + }, + ); + const fetchRemoteShared: GitVcsDriver.GitVcsDriver["Service"]["fetchRemote"] = (input) => + Cache.get( + fetchRemoteInFlight, + new FetchRemoteKey({ + cwd: input.cwd, + remoteName: input.remoteName, + refName: input.refName, + }), + ); const resolveRemoteTrackingCommit: GitVcsDriver.GitVcsDriver["Service"]["resolveRemoteTrackingCommit"] = Effect.fn("resolveRemoteTrackingCommit")(function* (input) { @@ -3861,21 +3898,29 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* if (input.oldBranch === input.newBranch) { return { branch: input.newBranch }; } - const targetBranch = input.exactName - ? input.newBranch - : yield* resolveAvailableBranchName(input.cwd, input.newBranch); + // `git branch -m` moves the branch's config section after the ref, so a + // lost config lock cannot be retried. The lock also keeps concurrent + // renames to one name from picking the same free suffix. + return yield* configLocks.withLock( + yield* resolveGitCommonDir(input.cwd), + Effect.gen(function* () { + const targetBranch = input.exactName + ? input.newBranch + : yield* resolveAvailableBranchName(input.cwd, input.newBranch); - yield* executeGit( - "GitVcsDriver.renameBranch", - input.cwd, - ["branch", "-m", "--", input.oldBranch, targetBranch], - { - timeoutMs: 10_000, - fallbackErrorDetail: "git branch rename failed", - }, - ); + yield* executeGit( + "GitVcsDriver.renameBranch", + input.cwd, + ["branch", "-m", "--", input.oldBranch, targetBranch], + { + timeoutMs: 10_000, + fallbackErrorDetail: "git branch rename failed", + }, + ); - return { branch: targetBranch }; + return { branch: targetBranch }; + }), + ); }); const switchRef: GitVcsDriver.GitVcsDriver["Service"]["switchRef"] = Effect.fn("switchRef")( @@ -4076,7 +4121,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* ensureRemote: (input) => withListRefsInvalidation(input.cwd, ensureRemote(input)), resolvePrimaryRemoteName, resolveDefaultBranchName, - fetchRemote: (input) => withListRefsInvalidation(input.cwd, fetchRemote(input)), + fetchRemote: (input) => withListRefsInvalidation(input.cwd, fetchRemoteShared(input)), remoteExists, remoteBranchExists, resolveRemoteTrackingCommit,