Overview
The MergeQueue manages serial integration of worker branches into the main branch. It implements priority-based ordering, automatic conflict retry with rebase, and health tracking.
Location: packages/orchestrator/src/merge-queue.ts
Class: MergeQueue
Constructor
new MergeQueue(
config: {
mergeStrategy: MergeStrategy
mainBranch: string
repoPath: string
gitMutex?: GitMutex
maxConflictRetries?: number
},
deps?: Partial<MergeQueueDeps>
)
config.mergeStrategy
'fast-forward' | 'rebase' | 'merge-commit'
required
Primary merge strategy (default: ‘rebase’)
Target branch name (default: ‘main’)
Optional mutex to serialize git operations
config.maxConflictRetries
Max retries before escalating conflict to fix task
Core Methods
enqueue()
Adds a branch to the merge queue.
enqueue(branch: string, priority: number = 5): void
Branch name to merge (e.g., “worker/task-001-auth”)
Lower = higher priority. Fix tasks use priority 1.
- Skips if branch already merged
- Skips if branch already in queue
Sorting:
queue.sort((a, b) =>
a.priority !== b.priority
? a.priority - b.priority // Primary: priority
: a.enqueuedAt - b.enqueuedAt // Secondary: FIFO
)
mergeBranch()
Attempts to merge a single branch.
async mergeBranch(branch: string): Promise<MergeQueueResult>
MergeQueueResult
interface MergeQueueResult {
success: boolean
status: 'merged' | 'skipped' | 'failed' | 'conflict'
branch: string
message: string
conflicts?: string[] // Conflicting file paths
}
- Acquire git mutex (if configured)
- Clean state: abort any in-progress rebase/merge, reset hard, checkout main
- Fetch branch:
git fetch origin <branch>
- Checkout main: Ensure clean working tree
- Merge:
git merge origin/<branch> with configured strategy
- Fallback: If rebase/fast-forward fails, retry with merge-commit
- Push:
git push origin main
- Cleanup: Delete remote branch
- Release mutex
Conflict Handling:
If merge results in conflict:
if (result.conflicted) {
const retries = this.retryCount.get(branch) ?? 0
if (retries < maxConflictRetries) {
// Rebase branch onto latest main
await rebaseBranch(branch, mainBranch)
// Re-enqueue with highest priority
this.enqueue(branch, 1)
this.retryCount.set(branch, retries + 1)
return { status: 'skipped', message: 'Conflict retry ...' }
}
// Retries exhausted → fire onConflict callback
for (const cb of this.conflictCallbacks) {
cb({ branch, conflictingFiles: conflicts })
}
return { status: 'conflict', conflicts }
}
processQueue()
Processes all queued branches sequentially.
async processQueue(): Promise<MergeQueueResult[]>
startBackground()
Starts background merge loop.
startBackground(intervalMs: number = 5_000): void
Tick interval in milliseconds
stopBackground() is called.
stopBackground()
Stops the background merge loop.
Merge Strategies
Rebase (default)
Git commands:
git fetch origin <branch>
git checkout main
git rebase origin/<branch>
git push origin main
Fast-Forward
mergeStrategy: 'fast-forward'
git fetch origin <branch>
git checkout main
git merge --ff-only origin/<branch>
git push origin main
Merge Commit
mergeStrategy: 'merge-commit'
git fetch origin <branch>
git checkout main
git merge --no-ff origin/<branch>
git push origin main
Conflict Retry with Rebase
Before re-queueing a conflicting branch:
try {
// Create local tracking branch
await git(['checkout', '-b', localBranch, `origin/${branch}`])
// Rebase onto latest main
const rebaseResult = await rebaseBranch(localBranch, mainBranch)
if (rebaseResult.success) {
// Force-push rebased branch
await git(['push', 'origin', `${localBranch}:${branch}`, '--force'])
logger.info('Rebased branch onto latest main before retry')
}
// Cleanup local branch
await git(['checkout', mainBranch])
await git(['branch', '-D', localBranch])
} catch (error) {
// Best-effort — restore clean state and continue
await ensureCleanState(mainBranch)
}
Statistics
getMergeStats()
getMergeStats(): MergeStats
interface MergeStats {
totalMerged: number // Successful merges
totalSkipped: number // Skipped (already merged, etc.)
totalFailed: number // Failed merges
totalConflicts: number // Conflicts (after retries exhausted)
}
isBranchMerged()
isBranchMerged(branch: string): boolean
getQueueLength()
Returns current queue depth.
resetRetryCount()
resetRetryCount(branch: string): void
Callbacks
onMergeResult()
onMergeResult(callback: (result: MergeQueueResult) => void): void
onConflict()
onConflict(callback: (info: MergeConflictInfo) => void): void
interface MergeConflictInfo {
branch: string
conflictingFiles: string[]
}
State Management
ensureCleanState()
Nuclear git cleanup — best-effort, never throws.
private async ensureCleanState(mainBranch: string): Promise<void>
git rebase --abort (best-effort)git merge --abort (best-effort)git reset --hard HEADgit clean -fd- Delete all
retry-rebase-* temp branches
git checkout <mainBranch>
Called before each merge and on errors to prevent git state corruption.
Usage Example
import { MergeQueue } from '@longshot/orchestrator'
const mergeQueue = new MergeQueue({
mergeStrategy: 'rebase',
mainBranch: 'main',
repoPath: './target-repo',
maxConflictRetries: 2
})
// Register callbacks
mergeQueue.onMergeResult((result) => {
console.log(`Merge ${result.branch}: ${result.status}`)
if (result.status === 'conflict') {
console.log(`Conflicts: ${result.conflicts?.join(', ')}`)
}
})
mergeQueue.onConflict(({ branch, conflictingFiles }) => {
console.log(`Creating fix task for ${branch}`)
console.log(`Conflicting files: ${conflictingFiles.join(', ')}`)
// Create and inject conflict-resolution task
})
// Enqueue branches
mergeQueue.enqueue('worker/task-001-auth', 5)
mergeQueue.enqueue('worker/task-002-api', 3)
// Start background processing
mergeQueue.startBackground(5_000)
// Or process synchronously
const results = await mergeQueue.processQueue()
console.log(`Merged ${results.filter(r => r.success).length} branches`)
// Get statistics
const stats = mergeQueue.getMergeStats()
console.log(`Success rate: ${stats.totalMerged}/${stats.totalMerged + stats.totalConflicts}`)
Dependency Injection
interface MergeQueueDeps {
execFileAsync: typeof execFileAsync
checkoutBranch: typeof checkoutBranch
mergeBranch: typeof mergeBranch
rebaseBranch: typeof rebaseBranch
now: () => number
setTimeout: typeof setTimeout
clearTimeout: typeof clearTimeout
}