“上周五我们仓库里积压了22个合并请求。其中一个已经等了三天因为CI一直在一片混沌中变红——你合一个进去它把另一个顶坏了你退一个出来又冒出新冲突。我觉得必须做点什么。于是就有了这套叫Gas Town Refinery的小工具一个按Bors风格设计的合并队列专门解决大批量PR合并时的顺序、冲突和CI验证问题。”不卖关子Gas Town Refinery的思路其实不复杂所有待合并的PR不是直接推入主分支而是先按顺序进入一个队列每满一批比如5个就把这一批合并到一个临时staging分支上全体跑一遍CI。只有整个批次通过测试主分支才会被快进到这个staging分叉点上。如果有人破坏了这个批次就把肇事者挑出来踢出队列剩下的人重新组队。听起来像自助餐厅传菜电梯一次端五盘菜菜没泼汤整个饭局才允许继续。这篇文章面向两类人一类是在维护中大型Git仓库、被“先合谁后合谁”折磨得够呛的基础设施开发另一类是单纯想看一眼合并队列是怎么用代码落地的。如果你只是听说过Bors那正好这里会把它掰开揉碎讲清楚。1. 项目背景与整体设计思路1.1 手动合并的致命循环我先还原一下那段混乱的日子。一个十几人的团队每天产生20个左右PR。管理员每天下午的做法是看哪个PR标题最急手动合并合并完CI挂掉赶紧再手动挑一个可能影响更小的PR撤销主分支变红几个开发者同时来喊“我改的文件和谁冲突了”。这本质上是整个仓库没有一个统一的“合并吞吐节奏”每次合并都在赌。Bors风格的合并队列之所以好用是因为它强制了这样一个原则主分支只能接受已经被完整验证过的合并结果。你不能把一个未经组合验证的PR直接塞进主干。所有PR都必须先进入一个受控队列队列把它们挨个合并到staging分支在那个临时分支上跑同一套CI。过了主干快进不过退回去重新排队。这样一来主分支永远是“某个经过真实合并、真实测试”的状态而不是“若干单独测试通过的PR的偶然叠加”。Gas Town Refinery落地的就是这个原则只不过我把适用范围压缩到了我们内部的Git自建仓库并用尽量少的依赖实现它。1.2 批处理带来的吞吐与风险平衡为什么选择批量合并而不是逐条合并逐条最安全但最慢。我们的CI流水线跑一次全量构建需要8分钟以前每次合一个PR要等8分钟一天只能处理少量PR。批量合并可以一次性跑完5个PR的组合验证时间几乎仍是8分钟。如果有冲突那恰好说明这两个PR不适合在同一批出现把它们拆开就好。批量合并真正的风险在于“组内互炸”A和B单独合入都能过但合在一起会触发编译错误。这是停掉手动合并之后才能暴露出来的问题也正是我们想要的——因为如果这两个PR分别在不同批次合入间隔太久可能谁都没有意识到它们之间有隐藏的语义冲突。批量验证实际上提高了团队对代码整体一致性的敏感度。所以批量不是一个拍脑袋的优化项而是从“时间成本”和“组合覆盖深度”两头看的必然选择。1.3 Gas Town Refinery的整体架构与选型这个工具从零开始我定了几条约束不引入需要部署额外常驻服务的东西选Python标准库加上调用本地git命令来完成核心逻辑。队列状态持久化用SQLite因为进程重启后队列不能丢不需要Redis这种重量级选手。与CI的交互尽量简单从本地轮询CI服务状态接口而不是做webhook回调这样最小可行版本能更快跑起来。整个系统分四个模块CLI入口解析配置文件启动后台守护进程。队列服务维护待合并PR的有序序列和生命周期。Git操作封装所有分支创建、合并、快进等统一走本地git子进程。CI轮询器定期查询远端CI状态判断批次是否通过。事实证明这种“轻依赖、本地优先”的选型在内部场景里非常稳。部署的时候只需要一台能访问仓库和CI接口的小机器就行不是每个人都要装一堆node_modules。2. 合并队列的数据结构与批次算法2.1 队列条目的状态机设计每一个进入Gas Town Refinery的PR会经历下面这几个状态QUEUED - BUIDING - TESTING - MERGED \- FAILED - RETRY - ...第一个版本我用了简单的字典列表硬撑后来发现一旦队列里超过10个PR状态很容易乱。改成SQLite后每条记录就是一行查询和回滚都方便。核心表结构大概长这样字段类型说明pr_numINTEGER主键PR编号source_branchTEXT来源分支target_branchTEXT目标分支目前固定为主分支statusTEXT当前状态attemptsINTEGER尝试合并次数staged_commitTEXT最近一次打包时的HEADcreated_atREAL入队时间为什么要把staged_commit也存下来因为一旦staging分支测试通过我要确保快进前主分支没有被别人动过。保存打包当时的commit哈希之后对比远端主分支的HEAD如果没变才能做fast-forward这是避免竞态的兜底。2.2 批次拆分的核心标准冲突检测既然选择了批量合并“哪些PR可以安全放进同一个批次”就成为最重要的算法问题。我试过两个办法。第一个是直接用git merge --no-commit做真实试合并冲突就回退。这种方法准确但是每次试合并都会修改工作区如果一次扫描10个PR工作区会被反复重置速度还行但比较“伤仓库”。第二个是优先用git merge-tree做快速感知。它能在不碰工作区和索引的情况下判断两个分支合并是否冲突。实测下来对于简单的文件冲突已经很准。不过要注意它的退出码不总是可信某些更隐蔽的语义冲突它发现不了。所以我的策略是先用merge-tree做粗筛真正加入批次时仍会进行一次干净的试合并。批次构建的伪代码大致是这样def build_batch(pending_prs, max_batch5): selected [] base latest_target_commit() for pr in pending_prs: if len(selected) max_batch: break # 先把当前待选分支拉到与目标分支一致再做试合并 run_git([fetch, origin, pr.source_branch]) run_git([checkout, -B, candidate-base, base]) status run_git( [merge, --no-ff, --no-commit, pr.source_branch], checkFalse ) if status.returncode ! 0: reset_worktree() continue # 不是从头到尾都失败就把这个PR剔出当前批次 selected.append(pr) base run_git([rev-parse, HEAD]) # 当前试点 reset_worktree() return selected一个细节如果第一个PR就冲突那说明它和目标分支之间有硬冲突。这时候不应该把它直接标记失败因为它可能只是暂时落后于主分支。正确做法是把这个PR留在队列头部等主分支上的同名字处理完或者等它rebase后再试一次。2.3 staging分支的建立与动作约束一旦选出了一个批次Gas Town Refinery会从当前主分支最新HEAD创建一个新的临时分支命名成staging-timestamp。然后按顺序用普通merge合入这5个PR。注意这里我不会用squash因为团队需要保留PR粒度出问题时也能快速剔除某一个。staging分支创建后马上push到远端远端CI会针对这个分支跑流水线。这里需要一个关键配置CI要监听staging-*分支而不是只监听主分支。否则你push上去根本没反应。CI跑完后如果状态是successGas Town Refinery会先去fetch远端主分支的最新HEAD确认和我们创建staging分支时的base一致如果一致就git merge --ff-only staging-*。如果中间主分支被人动了宁可把批次拆了重来也不能直接快进因为staging分支里可能已经包含了旧的主分支基线会丢失远程更新。3. 实操过程Gas Town Refinery的关键模块实现3.1 环境准备与目录结构实现这个工具不需要重型环境。我用的是Git 2.30支持git merge-tree合理退出码Python 3.9可访问目标和CI接口的HTTP凭据目录结构按模块拆方便后面扩展gas_town_refinery/ ├── config.yaml # 仓库地址、CI接口、批量大小等 ├── refinery/ │ ├── __init__.py │ ├── queue.py # SQLite持久化 队列状态机 │ ├── git_ops.py # git命令封装 │ ├── batch.py # 批次构建和合并 │ ├── ci_watch.py # CI状态轮询 │ └── cli.py # 命令行入口 └── tests/ ├── test_queue.py └── test_batch.py我刻意没有用像PyGithub这样的SDK因为自建仓库的API格式我们更熟悉直接用HTTP轮询反而灵活。git操作全部走subprocess虽然很多人觉得繁琐但你能精确控制每一步的退出码和输出这对判断合并状态非常关键。3.2 队列服务核心代码queue.py里我用一个轻量的MergeQueue类封装SQLite操作。关键点是所有涉及状态变更的方法都要加锁因为守护进程里轮询CI的主循环和CLI的新PR入队可能同时发生。import sqlite3 import threading import time from contextlib import contextmanager class MergeQueue: def __init__(self, db_path): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.lock threading.RLock() self._init_db() def _init_db(self): with self._cursor() as cur: cur.execute( CREATE TABLE IF NOT EXISTS queue ( pr_num INTEGER PRIMARY KEY, source_branch TEXT NOT NULL, target_branch TEXT NOT NULL, status TEXT NOT NULL DEFAULT QUEUED, attempts INTEGER DEFAULT 0, staged_commit TEXT, created_at REAL ) ) contextmanager def _cursor(self): with self.lock: cur self.conn.cursor() try: yield cur self.conn.commit() finally: cur.close() def enqueue(self, pr_num, source_branch, target_branch): with self._cursor() as cur: cur.execute( INSERT OR IGNORE INTO queue VALUES (?,?,?, QUEUED, 0, NULL, ?), (pr_num, source_branch, target_branch, time.time()) ) def get_pending(self, limit100): with self._cursor() as cur: rows cur.execute( SELECT pr_num, source_branch, target_branch FROM queue WHERE status QUEUED ORDER BY created_at LIMIT ?, (limit,) ).fetchall() return [dict(r) for r in rows] def set_status(self, pr_num, status, staged_commitNone): with self._cursor() as cur: cur.execute( UPDATE queue SET status?, staged_commit? WHERE pr_num?, (status, staged_commit, pr_num) )有一个容易踩的坑insert or ignore会让重复PR静默失败。我后来在enqueue里先查一次如果已存在就更新source_branch因为开发者通常会在原PR上继续push新commit队列里存的应该是“最新代码”而不是第一次入队时的快照。3.3 批次算法与冲突检测实现batch.py里最核心的部分是build_and_prepare_batch。它要做四件事取pending PR、试合并选批次、创建staging分支、把PR标记为BUILDING。实际代码比伪代码多了一层保护每次试合并前都一定要确保本地工作区是干净的。我吃过一次亏本地有未提交的改动结果试合并时把工作区搞脏了CI stage又被push到远端最后排查了半天才发现是本地的锅。所以我加了git diff --quiet检查不干净直接拒绝启动批次构建。def build_and_prepare_batch(queue, max_batch5): pending queue.get_pending() if not pending: return None if not is_worktree_clean(): raise RuntimeError(工作区有未提交改动拒绝构建批次) base get_remote_head(origin, main) selected [] tmp_branch fstaging-{int(time.time())} run_git([checkout, -B, tmp_branch, base]) for pr in pending: if len(selected) max_batch: break run_git([fetch, origin, pr[source_branch]]) result run_git( [merge, --no-ff, --no-commit, forigin/{pr[source_branch]}], checkFalse ) if result.returncode ! 0: run_git([reset, --hard, base]) continue selected.append(pr[pr_num]) new_head run_git([rev-parse, HEAD]).strip() run_git([reset, --hard, base]) # 准备下一个试合并 base new_head if not selected: run_git([checkout, main]) run_git([branch, -D, tmp_branch]) return None # 由于上面反复reset我们实际没有真的合并出结果因此需要重新合并一次 run_git([checkout, -B, tmp_branch, base]) for pr_num in selected: pr_info queue.get_pr(pr_num) run_git([merge, --no-ff, pr_info[source_branch]]) run_git([push, origin, frefs/heads/{tmp_branch}]) for pr_num in selected: queue.set_status(pr_num, BUILDING, staged_commitget_remote_head(origin, tmp_branch)) return tmp_branch有的读者会问为什么试合并完了还要重新合并一遍因为试合并过程中每加入一个PR我都会reset回基线所以最终staging分支上必须再真刀真枪合并一次保证最后的树和试合并时选择的“成功路径”一致。这一步虽然冗余但能彻底避免合并顺序或中间态带来的误差。3.4 CI状态监听与最终快进ci_watch.py的任务很简单每隔10秒查询一次CI接口直到这个staging分支的状态不是pending。我这里用的是内部CI的HTTP接口返回json{branch: staging-123, state: success, url: /jobs/123}轮询器只认三个状态pending、success、failure。连续60次查询还是pending就标记超时。一旦成功调用fast_forward_main(staging_branch)先fetch远端main再拿远端main的HEAD和我之前记录staging分支的base对比。如果一致执行网络快进git fetch origin staging-123:staging-123 git checkout main git merge --ff-only refs/heads/staging-123 git push origin main如果主分支HEAD不一致说明这10分钟里有人直接改了主分支比如紧急hotfix这时候我不会强行覆盖。我会丢弃这个批次把所有涉及PR重新置回QUEUED状态让它们和新代码再排一轮。代价是多跑一轮CI但安全性高。4. 常见问题与排查技巧实录4.1 CI失败时怎么快速定位还不是唯一解决方案批次CI失败第一反应总是“谁搞的”。直接看CI日志里哪个测试挂了再根据测试涉及的文件映射到PR。但现实里有些错误是组合性的单看日志很难归因。我用的可靠方案是二分排除法。举个例子批次里有5个PR第1次试验发现编译错误。接下来我把5个PR分成前2个和后3个分别构建两个staging分支哪个分支失败了就进一步拆分那个分支内的PR。最多两三轮就能定位到单个PR。如果两次分组都失败说明失败可能与目标分支最新代码有关就先把批次里所有PR踢回队列人工干预一次。二分排除法看起来很机械但它能最快速地发现问题。我建议把这个操作也变成Gas Town Refinery的子命令省得手动敲一堆git命令。4.2 staging分支没有触发CI最常见的配置问题。CI系统默认只监听main分支你push一个staging分支上去根本不会触发流水线。这其实不是Gas Town Refinery的bug而是你还没有告诉CI“这是个候选发布线”。解决办法是在CI配置中把staging-*加入分支模式。比如某内部CI平台的配置长这样triggers: branches: - main - staging-*还有一个权限问题如果用的是GitHub Actionspush staging分支的账户必须至少拥有该仓库的写入权限。如果你的bot账号权限不够push会被拒。可以在bot的令牌权限里放开但注意不要把令牌写死在代码里。4.3 本地区分树的冲突判断远端不一定我在第2节说过merge-tree判断不一定全准。实际中有一种情况本地试合并成功staging push到远端后远端仓库的CI编译却失败了。原因通常是远端CI的构建环境里还会拉取一些私有依赖或者用了不同平台上的路径大小写敏感设置。这类问题最坑因为它不体现在git冲突里。我的应对是一旦出现这种“非结构冲突”的CI失败先把PR从批次中单独剔除再单独跑一个最小批次测试。如果单PR也失败那大概率是环境类问题例如CI构建脚本对Python版本敏感。这时候谁合并谁都一样要修的是CI环境而不是队列。4.4 快进合并前总有急事插队团队成员会在队列运行时直接改主分支发紧急修复这在成熟的Bors流程里是不允许的但小团队很难强制。Gas Town Refinery在这一块做了折中快进前不仅比较主分支HEAD还会检查主分支基于我们staging分支base之后有没有新的commit。有就放弃本次快进并把批次拆了重排。有些人觉得这样太浪费其实可以反过来利用设置一个“批次数上限”如果某个PR的尝试次数超过3次就把它标记为requires_manual_review不再自动处理。让真人介入避免一个顽固PR反复消耗CI资源。问题症状解决方案批次CI失败无法找出肇事PR二分排除法最小化到单个PRstaging不触发CIpush后没有任何构建任务检查CI分支匹配和令牌权限试合并通过但远端失败平台差异导致构建问题单独跑该PR构建排除环境差异主分支被插队快进时HEAD不一致自动重置批次PR回队列重排顽固PR反复失败浪费CI资源超过重试次数后移出自动队列标记人工处理5. 后续扩展方向与个人体会如果要把Gas Town Refinery做得更贴近生产级我会加这三个能力自动rebase入队时如果和最新主分支落后太多先rebase再进队减少后续冲突。多目标分支支持目前只处理单一主分支但发布分支和开发分支都要排队时需要区分不同的队列。回滚编排主分支快进后如果线上出现问题可以自动回滚到上一个已测试版本而不用让运维手工操作。但我个人的体会是自动合并工具最重要的不是功能多而是它能建立团队的“节奏感”。一旦大家知道有队列存在就默认所有合并都必须经过验证减少了很多象征性的“我先合进去再说”。我建议你从最小实现开始先拿两三个人的小仓库跑一两周观察队列空转率、批次失败率再决定要不要把BATCH_SIZE调大。如果你问我“5个一合和10个一合哪个好”我只能说先跑起来看数据在我的项目里5个一合时的失败率最可控10个一合偶尔会因为隐性冲突炸掉整批。最后分享一个小习惯每次创建staging分支后我都会在PR列表里贴一条评论注明“该PR已进入批次staging-xxx预计CI耗时约10分钟”。这大大减少了开发者来问“我那单什么时候合”的次数也算这个工具给团队的意外福利。