Serialize per-worktree git mutations to fix rebase race

Introduce operation locking for each worktree to prevent concurrent
mutations (like rebase) from interfering with each other. Ensures
rebasing a linked worktree doesn't affect the source worktree state.
Add SIGKILL fallback if process termination barriers cannot verify
tree termination.
This commit is contained in:
Jinjing
2026-08-24 11:35:14 -07:00
parent ce743a4fd0
commit 21cdce5cf7
10 changed files with 268 additions and 5 deletions
+11
View File
@@ -1,6 +1,7 @@
import { randomUUID } from 'node:crypto'
import { normalizeGitErrorMessage } from '../../shared/git-remote-error'
import { isNoWriteFetchHeadUnsupportedError } from '../../shared/git-fetch-head-capability'
import { runWithGitWorktreeOperationLock } from '../../shared/git-worktree-operation-lock'
import {
REBASE_SOURCE_FETCH_TIMEOUT_MS,
resolveGitRemoteRebaseSource
@@ -15,6 +16,16 @@ export async function gitPullRebaseFromBase(
worktreePath: string,
baseRef: string,
options: GitRuntimeOptions = {}
): Promise<void> {
await runWithGitWorktreeOperationLock(worktreePath, options.signal, () =>
gitPullRebaseFromBaseUnlocked(worktreePath, baseRef, options)
)
}
async function gitPullRebaseFromBaseUnlocked(
worktreePath: string,
baseRef: string,
options: GitRuntimeOptions
): Promise<void> {
await runWithGitReadCacheInvalidation(async () => {
const operationOptions = {
+86
View File
@@ -241,6 +241,92 @@ describe('GitHandler', () => {
}
}, 15_000)
it('rebases the selected linked worktree without moving the source worktree', async () => {
const bareDir = mkdtempSync(path.join(tmpdir(), 'relay-git-linked-rebase-bare-'))
const producerParent = mkdtempSync(path.join(tmpdir(), 'relay-git-linked-rebase-producer-'))
const producerDir = path.join(producerParent, 'repo')
const targetParent = mkdtempSync(path.join(tmpdir(), 'relay-git-linked-rebase-target-'))
const targetDir = path.join(targetParent, 'feature')
try {
execFileSync('git', ['init', '--bare'], { cwd: bareDir, stdio: 'pipe' })
gitInit(tmpDir)
writeFileSync(path.join(tmpDir, 'base.txt'), 'base')
gitCommit(tmpDir, 'base')
const baseBranch = execFileSync('git', ['branch', '--show-current'], {
cwd: tmpDir,
encoding: 'utf-8'
}).trim()
execFileSync('git', ['remote', 'add', 'origin', bareDir], { cwd: tmpDir, stdio: 'pipe' })
execFileSync('git', ['push', '--set-upstream', 'origin', baseBranch], {
cwd: tmpDir,
stdio: 'pipe'
})
execFileSync('git', ['worktree', 'add', '-b', 'feature', targetDir, 'HEAD'], {
cwd: tmpDir,
stdio: 'pipe'
})
writeFileSync(path.join(targetDir, 'topic.txt'), 'topic')
gitCommit(targetDir, 'topic')
writeFileSync(path.join(tmpDir, 'source-dirty.txt'), 'leave me alone')
const sourceHeadBefore = execFileSync('git', ['rev-parse', 'HEAD'], {
cwd: tmpDir,
encoding: 'utf-8'
}).trim()
const sourceStatusBefore = execFileSync('git', ['status', '--short'], {
cwd: tmpDir,
encoding: 'utf-8'
})
execFileSync('git', ['clone', bareDir, producerDir], { stdio: 'pipe' })
execFileSync('git', ['config', 'user.email', 'test@test.com'], {
cwd: producerDir,
stdio: 'pipe'
})
execFileSync('git', ['config', 'user.name', 'Test'], { cwd: producerDir, stdio: 'pipe' })
execFileSync('git', ['checkout', baseBranch], { cwd: producerDir, stdio: 'pipe' })
writeFileSync(path.join(producerDir, 'latest.txt'), 'latest')
gitCommit(producerDir, 'latest base')
const latestBaseOid = execFileSync('git', ['rev-parse', 'HEAD'], {
cwd: producerDir,
encoding: 'utf-8'
}).trim()
execFileSync('git', ['push', 'origin', baseBranch], { cwd: producerDir, stdio: 'pipe' })
await dispatcher.callRequest('git.rebaseFromBase', {
worktreePath: targetDir,
baseRef: `origin/${baseBranch}`
})
expect(execFileSync('git', ['rev-parse', 'HEAD'], { cwd: tmpDir }).toString().trim()).toBe(
sourceHeadBefore
)
expect(execFileSync('git', ['status', '--short'], { cwd: tmpDir, encoding: 'utf-8' })).toBe(
sourceStatusBefore
)
expect(
execFileSync('git', ['rev-parse', 'HEAD~1'], { cwd: targetDir, encoding: 'utf-8' }).trim()
).toBe(latestBaseOid)
expect(
execFileSync('git', ['reflog', '-1', '--format=%gs'], {
cwd: targetDir,
encoding: 'utf-8'
})
).toContain('rebase (finish)')
expect(
execFileSync('git', ['for-each-ref', '--format=%(refname)', 'refs/orca/rebase'], {
cwd: targetDir,
encoding: 'utf-8'
}).trim()
).toBe('')
} finally {
await Promise.all([
fs.rm(bareDir, { recursive: true, force: true }),
fs.rm(producerParent, { recursive: true, force: true }),
fs.rm(targetParent, { recursive: true, force: true })
])
}
}, 15_000)
it('fast-forwards an unborn branch from the selected remote base', async () => {
const bareDir = mkdtempSync(path.join(tmpdir(), 'relay-git-unborn-bare-'))
const producerDir = mkdtempSync(path.join(tmpdir(), 'relay-git-unborn-producer-'))
+7
View File
@@ -62,6 +62,7 @@ import { upstreamOnlyCommitsArePatchEquivalent } from '../shared/git-upstream-st
import { assertGitPushTargetShape } from '../shared/git-push-target-validation'
import { getPublishTargetStatus, type GitCommandRunner } from '../shared/git-publish-target-status'
import { isNoWriteFetchHeadUnsupportedError } from '../shared/git-fetch-head-capability'
import { runWithGitWorktreeOperationLock } from '../shared/git-worktree-operation-lock'
import { resolveGitFetchHeadCommand, runWithGitFetchHeadLock } from '../shared/git-fetch-head-lock'
import {
REBASE_FROM_BASE_OPERATION_TIMEOUT_MS,
@@ -1190,6 +1191,12 @@ export class GitHandler {
}
private async rebaseFromBase(params: Record<string, unknown>, context?: RequestContext) {
return runWithGitWorktreeOperationLock(params.worktreePath as string, context?.signal, () =>
this.runRebaseFromBase(params, context)
)
}
private async runRebaseFromBase(params: Record<string, unknown>, context?: RequestContext) {
this.clearGitMutationReadCaches()
const worktreePath = params.worktreePath as string
const baseRef = params.baseRef as string
@@ -22,6 +22,7 @@ import {
type RuntimeEnvironmentCallRequest
} from './runtime-compatibility-test-fixture'
import { clearRuntimeCompatibilityCacheForTests } from './runtime-rpc-client'
import { REBASE_FROM_BASE_RPC_TIMEOUT_MS } from '../../../shared/git-rebase-source'
const gitStatus = vi.fn()
const gitCancelStatus = vi.fn()
@@ -616,7 +617,7 @@ describe('runtime git client', () => {
selector: 'env-1',
method: 'git.rebaseFromBase',
params: { worktree: 'id:wt-1', baseRef: 'origin/main' },
timeoutMs: 30_000
timeoutMs: REBASE_FROM_BASE_RPC_TIMEOUT_MS
})
})
@@ -23,6 +23,7 @@ import type { HostedReviewProvider } from '../../../shared/hosted-review'
import type { ResolvedSourceControlAiGenerationParams } from '../../../shared/source-control-ai'
import { getCommitMessageModelDiscoveryHostKeyForScope } from '../../../shared/commit-message-host-key'
import type { GitHistoryOptions, GitHistoryResult } from '../../../shared/git-history'
import { REBASE_FROM_BASE_RPC_TIMEOUT_MS } from '../../../shared/git-rebase-source'
import { getRepoIdFromWorktreeId, splitWorktreeIdForFilesystem } from '../../../shared/worktree/id'
import { callRuntimeRpc, getActiveRuntimeTarget } from './runtime-rpc-client'
import { toRuntimeWorktreeSelector } from './runtime-worktree-selector'
@@ -548,7 +549,7 @@ export async function rebaseRuntimeGitFromBase(
target,
'git.rebaseFromBase',
{ worktree: toRuntimeWorktreeSelector(context.worktreeId), baseRef },
{ timeoutMs: 30_000 }
{ timeoutMs: REBASE_FROM_BASE_RPC_TIMEOUT_MS }
)
}
@@ -38,7 +38,7 @@ describe('runProcess termination failure', () => {
vi.clearAllMocks()
})
it('waits for root exit when bounded tree termination cannot be verified', async () => {
it('holds the result until the barrier deadline when tree termination cannot be verified', async () => {
const child = mockChild()
spawnMock.mockReturnValue(child)
const pending = runProcess({ program: 'git', timeoutMs: 10, terminationBarrier: true })
@@ -52,6 +52,8 @@ describe('runProcess termination failure', () => {
expect(settled).toBe(false)
child.emit('exit', null, 'SIGKILL')
expect(settled).toBe(false)
await vi.advanceTimersByTimeAsync(10_000)
await expect(pending).resolves.toMatchObject({ timedOut: true })
})
@@ -118,6 +120,28 @@ describe('runProcess termination failure', () => {
expect(settled).toBe(false)
child.emit('exit', null, 'SIGKILL')
await vi.advanceTimersByTimeAsync(10_000)
await rejection
})
it('kills the root when an object barrier cannot verify tree termination', async () => {
const child = mockChild()
spawnMock.mockReturnValue(child)
const pending = runProcess({
program: 'wsl.exe',
timeoutMs: 10,
terminationBarrier: {
signal: vi.fn().mockResolvedValue(false),
force: vi.fn().mockResolvedValue(false)
}
})
void pending.catch(() => {})
await vi.advanceTimersByTimeAsync(2_010)
expect(child.kill).toHaveBeenCalledWith('SIGKILL')
child.emit('exit', null, 'SIGKILL')
await vi.advanceTimersByTimeAsync(10_000)
await expect(pending).resolves.toMatchObject({ timedOut: true })
})
})
+16 -1
View File
@@ -198,6 +198,7 @@ export function runProcess(spec: ProcessSpec): Promise<ProcessResult> {
let deferredExit: { code: number | null; signal: NodeJS.Signals | null } | null = null
let deferredClose: { code: number | null; signal: NodeJS.Signals | null } | null = null
let deferredError: Error | null = null
let rootExitedBeforeBarrier = false
const settle = (act: () => void): void => {
if (settled) {
@@ -257,7 +258,7 @@ export function runProcess(spec: ProcessSpec): Promise<ProcessResult> {
const resolveBarrierIfSafe = (): void => {
const rootExit = deferredClose ?? deferredExit
if (barrierTerminationVerified || (barrierAttemptComplete && rootExit)) {
if (barrierTerminationVerified || (rootExitedBeforeBarrier && rootExit)) {
settleBarrierOutcome()
return
}
@@ -304,6 +305,11 @@ export function runProcess(spec: ProcessSpec): Promise<ProcessResult> {
([initialTerminated, forceTerminated]) => {
barrierAttemptComplete = true
barrierTerminationVerified = initialTerminated || forceTerminated
if (!barrierTerminationVerified) {
// The barrier never confirmed the tree died, so the root
// would otherwise outlive the abort or timeout.
terminate(child, 'SIGKILL')
}
resolveBarrierIfSafe()
}
)
@@ -323,6 +329,9 @@ export function runProcess(spec: ProcessSpec): Promise<ProcessResult> {
([_initialTerminated, forceTerminated]) => {
barrierAttemptComplete = true
barrierTerminationVerified = forceTerminated
if (!barrierTerminationVerified) {
terminate(child, 'SIGKILL')
}
resolveBarrierIfSafe()
}
)
@@ -360,12 +369,18 @@ export function runProcess(spec: ProcessSpec): Promise<ProcessResult> {
settle(() => reject(error))
})
child.once('exit', (code, signal) => {
if (!barrierStopping) {
rootExitedBeforeBarrier = true
}
deferredExit = { code, signal }
if (barrierStopping) {
resolveBarrierIfSafe()
}
})
child.once('close', (code, signal) => {
if (!barrierStopping) {
rootExitedBeforeBarrier = true
}
if (barrierStopping) {
deferredClose = { code, signal }
resolveBarrierIfSafe()
+2 -1
View File
@@ -1,7 +1,8 @@
// Why: a stalled remote must fail the rebase fetch, not hang the rebase; client and relay share one bound.
export const REBASE_SOURCE_FETCH_TIMEOUT_MS = 60_000
export const REBASE_FROM_BASE_OPERATION_TIMEOUT_MS = REBASE_SOURCE_FETCH_TIMEOUT_MS + 60_000
export const REBASE_FROM_BASE_RPC_TIMEOUT_MS = REBASE_FROM_BASE_OPERATION_TIMEOUT_MS + 5_000
// Include the process barrier's 2s grace plus its 10s unverified-tree deadline.
export const REBASE_FROM_BASE_RPC_TIMEOUT_MS = REBASE_FROM_BASE_OPERATION_TIMEOUT_MS + 15_000
export type GitCommandRunner = (args: string[]) => Promise<{ stdout: string }>
@@ -0,0 +1,48 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import { runWithGitWorktreeOperationLock } from './git-worktree-operation-lock'
describe('git worktree operation lock', () => {
let root: string | undefined
afterEach(async () => {
if (root) {
await rm(root, { recursive: true, force: true })
root = undefined
}
})
it('serializes mutations for the same linked worktree path', async () => {
root = await mkdtemp(join(tmpdir(), 'git-worktree-operation-lock-'))
const worktreePath = join(root, 'worktree')
const events: string[] = []
let releaseFirst!: () => void
let markFirstStarted!: () => void
const firstStarted = new Promise<void>((resolve) => {
markFirstStarted = resolve
})
const firstRelease = new Promise<void>((resolve) => {
releaseFirst = resolve
})
const first = runWithGitWorktreeOperationLock(worktreePath, undefined, async () => {
events.push('first:start')
markFirstStarted()
await firstRelease
events.push('first:end')
})
await firstStarted
const second = runWithGitWorktreeOperationLock(worktreePath, undefined, async () => {
events.push('second:start')
})
await Promise.resolve()
expect(events).toEqual(['first:start'])
releaseFirst()
await Promise.all([first, second])
expect(events).toEqual(['first:start', 'first:end', 'second:start'])
})
})
+69
View File
@@ -0,0 +1,69 @@
import { realpath } from 'node:fs/promises'
import { resolve } from 'node:path'
type OperationLane = {
tail: Promise<void>
release: () => void
}
const lanes = new Map<string, OperationLane>()
function abortError(): Error {
const error = new Error('The operation was aborted.')
error.name = 'AbortError'
return error
}
async function waitForPredecessor(
predecessor: Promise<void>,
signal: AbortSignal | undefined
): Promise<void> {
if (!signal) {
await predecessor.catch(() => undefined)
return
}
if (signal.aborted) {
throw abortError()
}
let rejectAbort!: (error: Error) => void
const aborted = new Promise<never>((_resolve, reject) => {
rejectAbort = reject
})
const onAbort = () => rejectAbort(abortError())
signal.addEventListener('abort', onAbort, { once: true })
try {
await Promise.race([predecessor.catch(() => undefined), aborted])
} finally {
signal.removeEventListener('abort', onAbort)
}
}
/** Serialize mutations that leave per-worktree state in progress (for example, rebase). */
export async function runWithGitWorktreeOperationLock<T>(
worktreePath: string,
signal: AbortSignal | undefined,
run: () => Promise<T>
): Promise<T> {
const key = await realpath(worktreePath).catch(() => resolve(worktreePath))
const predecessor = lanes.get(key)?.tail ?? Promise.resolve()
let release!: () => void
const current = new Promise<void>((resolveCurrent) => {
release = resolveCurrent
})
const lane = {
tail: predecessor.catch(() => undefined).then(() => current),
release
}
lanes.set(key, lane)
void lane.tail.then(() => {
if (lanes.get(key) === lane) {
lanes.delete(key)
}
})
try {
await waitForPredecessor(predecessor, signal)
return await run()
} finally {
lane.release()
}
}