MergeQueue
§ · Nested detail page

FIFO queue-lock 深度解析

PID-liveness FIFO lock 如何让 build 和 land 在本机跨 worktree 串行执行。

§ 00 · DETAIL

导入链与契约

锁的公共接口

QueueLock 的公共契约很小：src/lib/queue-lock.ts:43 定义 AcquireOptions，src/lib/queue-lock.ts:48 暴露 acquire/release/lane/held。这个接口克制，意味着 build-lock 和 land 可以共用一套互斥原语，而不用各自发明不同的排队语义。src/lib/queue-lock.ts:55 的 repoKey 又把锁绑定到 git common dir，不会让同一个 repo 的多个 worktree 锁分裂。

真正的设计点在 src/lib/queue-lock.ts:73 到 src/lib/queue-lock.ts:80：queueName 决定是 build 队列还是 land 队列，repo hash 决定作用域，tickets 子目录和 lock 文件分别承担排队和持锁状态。src/lib/queue-lock.ts:82 还把当前目录名作为 lane，方便等待日志显示是哪条 lane。

src/lib/queue-lock.ts:43 · L43-86TypeScript
43export interface AcquireOptions {
44  label?: string;
45  onWait?: (info: { ahead: number; holder: LockHolder | null }) => void;
46}
47
48export interface QueueLock {
49  acquire(options?: AcquireOptions): Promise<void>;
50  release(): void;
51  readonly lane: string;
52  readonly held: boolean;
53}
54
55function repoKey(): string {
56  try {
57    const commonDir = execSync("git rev-parse --git-common-dir", {
58      encoding: "utf8",
59    }).trim();
60    // Resolve to an absolute, worktree-independent path so every worktree of
61    // the same repo hashes to the same queue.
62    return execSync(`cd "${commonDir}" && pwd -P`, { encoding: "utf8" }).trim();
63  } catch {
64    return process.cwd();
65  }
66}
67
68/**
69 * Create a named FIFO lock. Each distinct `queueName` is an independent
70 * mutex — "build" and "land" never contend with each other even though
71 * they share this exact same code.
72 */
73export function createQueueLock(queueName: string): QueueLock {
74  const QUEUE_DIR = join(
75    tmpdir(),
76    `claude-code-merge-queue-${queueName}-queue-${createHash("sha1").update(repoKey()).digest("hex").slice(0, 12)}`,
77  );
78  const TICKETS_DIR = join(QUEUE_DIR, "tickets");
79  const LOCK_FILE = join(QUEUE_DIR, "lock");
80  mkdirSync(TICKETS_DIR, { recursive: true });
81
82  const lane = basename(process.cwd());
83  const ME = process.pid;
84  const TICKET_TS = Date.now();
85  const TICKET_NAME = `${TICKET_TS}-${ME}`;
86  const TICKET_FILE = join(TICKETS_DIR, TICKET_NAME);
§ 01 · DETAIL

核心实现逐段拆

公平队列

排队公平性来自 src/lib/queue-lock.ts:103 的 pruneDeadTickets 和 src/lib/queue-lock.ts:122 的排序。它按 timestamp 再按 pid 排序，所以不是“谁 poll 得快谁赢”。这比简单 while-lock-file-exists 更重要，因为多个 coding agent 往往会同时结束测试，公平队列能减少重复 rebase 和饥饿。

原子抢锁在 src/lib/queue-lock.ts:140。写临时文件后用 src/lib/queue-lock.ts:144 的 linkSync 建立 lock 文件：如果目标已存在，文件系统保证失败。失败后 src/lib/queue-lock.ts:154 读取 holder，并在 src/lib/queue-lock.ts:155 判断 PID 是否还活着；死 holder 会在 src/lib/queue-lock.ts:157 被回收。

src/lib/queue-lock.ts:88 · L88-127TypeScript
88  function alive(pid: number): boolean {
89    if (!pid || pid === ME) return pid === ME;
90    try {
91      process.kill(pid, 0);
92      return true;
93    } catch (e) {
94      return (e as NodeJS.ErrnoException).code === "EPERM"; // exists but owned by someone else
95    }
96  }
97
98  function pidOf(name: string): number {
99    const dash = name.lastIndexOf("-");
100    return dash === -1 ? 0 : Number(name.slice(dash + 1));
101  }
102
103  function pruneDeadTickets(): string[] {
104    let names: string[];
105    try {
106      names = readdirSync(TICKETS_DIR);
107    } catch {
108      return [];
109    }
110    const live: string[] = [];
111    for (const name of names) {
112      if (alive(pidOf(name))) {
113        live.push(name);
114      } else {
115        try {
116          unlinkSync(join(TICKETS_DIR, name));
117        } catch {
118          /* someone else cleaned it */
119        }
120      }
121    }
122    live.sort((a, b) => {
123      const [ta, pa] = a.split("-").map(Number) as [number, number];
124      const [tb, pb] = b.split("-").map(Number) as [number, number];
125      return ta - tb || pa - pb;
126    });
127    return live;
src/lib/queue-lock.ts:130 · L130-164TypeScript
130  function readLockHolder(): LockHolder | null {
131    try {
132      return JSON.parse(readFileSync(LOCK_FILE, "utf8")) as LockHolder;
133    } catch {
134      return null;
135    }
136  }
137
138  // Atomically take the lock via link() (fails if it already exists). Reclaim
139  // a lock whose holder is dead. Returns true iff we now hold it.
140  function tryTakeLock(info: LockHolder): boolean {
141    const tmp = `${LOCK_FILE}.${ME}.tmp`;
142    writeFileSync(tmp, JSON.stringify(info));
143    try {
144      linkSync(tmp, LOCK_FILE);
145      unlinkSync(tmp);
146      return true;
147    } catch (e) {
148      try {
149        unlinkSync(tmp);
150      } catch {
151        /* noop */
152      }
153      if ((e as NodeJS.ErrnoException).code !== "EEXIST") throw e;
154      const holder = readLockHolder();
155      if (!holder || !alive(holder.pid)) {
156        try {
157          unlinkSync(LOCK_FILE);
158        } catch {
159          /* another waiter beat us to it */
160        }
161      }
162      return false;
163    }
164  }
§ 02 · DETAIL

关键流程

获取与释放

acquire 的循环在 src/lib/queue-lock.ts:192 开始：先写 ticket，再每轮 prune，检查自己在队列中的位置。只有 src/lib/queue-lock.ts:201 满足 ahead <= 0 且 lockFree 时才尝试拿锁。成功后 src/lib/queue-lock.ts:205 删除自己的 ticket，说明 holder 与 waiter 身份分离，队列目录不会把已持锁者重复算作等待者。

release 也保持幂等：src/lib/queue-lock.ts:167 只在 held 时删除 lock，并验证 src/lib/queue-lock.ts:170 holder pid 是自己；随后 src/lib/queue-lock.ts:179 不管是否 held 都尝试删 ticket。这让正常退出和异常恢复走同一条清理路径。

ticket file → prune dead pids → oldest live ticket → linkSync lock → release/unlink
src/lib/queue-lock.ts:166 · L166-221TypeScript
166  let HOLD = false;
167  function release(): void {
168    if (HOLD) {
169      const holder = readLockHolder();
170      if (holder && holder.pid === ME) {
171        try {
172          unlinkSync(LOCK_FILE);
173        } catch {
174          /* already gone */
175        }
176      }
177      HOLD = false;
178    }
179    try {
180      unlinkSync(TICKET_FILE);
181    } catch {
182      /* already gone */
183    }
184  }
185
186  const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
187
188  /**
189   * Wait for and take the lock. `onWait({ahead, holder})` fires whenever the
190   * queue position changes, so callers can print progress.
191   */
192  async function acquire({ label, onWait }: AcquireOptions = {}): Promise<void> {
193    writeFileSync(TICKET_FILE, JSON.stringify({ pid: ME, lane, label, ts: TICKET_TS }));
194    let announced = -1;
195    for (;;) {
196      const queue = pruneDeadTickets();
197      const ahead = queue.indexOf(TICKET_NAME); // 0 = our turn
198      const holder = readLockHolder();
199      const lockFree = !holder || !alive(holder.pid);
200
201      if (ahead <= 0 && lockFree) {
202        if (tryTakeLock({ pid: ME, lane, label, ts: Date.now() })) {
203          HOLD = true;
204          try {
205            unlinkSync(TICKET_FILE);
206          } catch {
207            /* noop */
208          }
209          return;
210        }
211      }
212
213      if (ahead > 0 && ahead !== announced) {
214        announced = ahead;
215        onWait?.({ ahead, holder: null });
216      } else if (ahead <= 0 && holder && alive(holder.pid) && announced !== 0) {
217        announced = 0;
218        onWait?.({ ahead: 0, holder });
219      }
220      await sleep(POLL_MS);
221    }
§ 03 · DETAIL

边界与失败

测试直接覆盖了两个最关键性质。test/queue-lock.test.ts:32 启动四个真实 worker，test/queue-lock.test.ts:49 按开始时间排序，再用 test/queue-lock.test.ts:53 断言区间不重叠。崩溃恢复测试在 test/queue-lock.test.ts:59 开始，test/queue-lock.test.ts:75 用 kill -9 杀 holder，随后 test/queue-lock.test.ts:80 启动第二个 worker，证明没有永久死锁。

test/queue-lock.test.ts:32 · L32-88TypeScript
32test("mutual exclusion: four concurrent workers never overlap while holding the lock", async () => {
33  const dir = mkdtempSync(join(tmpdir(), "claude-code-merge-queue-lock-test-"));
34  const resultsFile = join(dir, "results.ndjson");
35  const queueName = `test-mutex-${process.pid}-${Date.now()}`;
36
37  const workers = [0, 1, 2, 3].map(() => spawnWorker(queueName, resultsFile, 150));
38  await Promise.all(workers.map(waitExit));
39
40  const events = readEvents(resultsFile);
41  const intervals = new Map<number, { start: number; end: number }>();
42  for (const e of events) {
43    const cur = intervals.get(e.pid) ?? { start: 0, end: 0 };
44    cur[e.event === "start" ? "start" : "end"] = e.t;
45    intervals.set(e.pid, cur);
46  }
47  assert.equal(intervals.size, 4, "expected all four workers to record start+end");
48
49  const sorted = [...intervals.values()].sort((a, b) => a.start - b.start);
50  for (let i = 1; i < sorted.length; i++) {
51    const prev = sorted[i - 1]!;
52    const cur = sorted[i]!;
53    assert.ok(cur.start >= prev.end, `worker holding [${prev.start},${prev.end}] overlapped worker starting at ${cur.start}`);
54  }
55
56  rmSync(dir, { recursive: true, force: true });
57});
58
59test("crash safety: a killed holder is reclaimed, not a permanent deadlock", async () => {
60  const dir = mkdtempSync(join(tmpdir(), "claude-code-merge-queue-lock-test-"));
61  const resultsFile = join(dir, "results.ndjson");
62  const queueName = `test-crash-${process.pid}-${Date.now()}`;
63
64  // Worker A holds the lock for far longer than the test's patience — it
65  // should never release it gracefully. It gets SIGKILLed instead.
66  const holder = spawnWorker(queueName, resultsFile, 60_000);
67
68  const deadline = Date.now() + 5_000;
69  while (Date.now() < deadline) {
70    if (existsSync(resultsFile) && readFileSync(resultsFile, "utf8").includes('"start"')) break;
71    await new Promise((r) => setTimeout(r, 50));
72  }
73  assert.ok(existsSync(resultsFile) && readFileSync(resultsFile, "utf8").includes('"start"'), "holder never acquired the lock");
74
75  execSync(`kill -9 ${holder.pid}`);
76
77  // Worker B should still be able to acquire the SAME queue — if the lock
78  // weren't reclaimed via PID-liveness, this would hang until the test
79  // runner's own timeout, not queue-lock's (there is no queue-lock timeout).
80  const second = spawnWorker(queueName, resultsFile, 10);
81  const code = await waitExit(second);
82  assert.equal(code, 0, "second worker should acquire and finish cleanly after the holder was killed");
83
84  const events = readEvents(resultsFile).filter((e) => e.pid === second.pid);
85  assert.equal(events.length, 2, "second worker should have recorded both start and end");
86
87  rmSync(dir, { recursive: true, force: true });
88});
§ 04 · DETAIL

关联模块

这个锁不是孤立模块。land 在 src/land.ts:78 创建名为 land 的队列，build-lock 在 src/build-lock.ts:1 使用同类语义，而 ephemeral resource 在 src/lib/ephemeral.ts:105 把 claim/use/release 模式迁移到测试资源。换句话说，项目把“进程活性判断”作为基础工程原语复用。

src/lib/ephemeral.ts:97 · L97-126TypeScript
97/**
98 * Run `fn` against a freshly created resource, prune any orphans left by a
99 * previous crashed run first, and always release the claim — including on
100 * a thrown error. If THIS process is killed before the `finally` runs, the
101 * resource isn't leaked forever: the next call to `withEphemeralResource`
102 * (in the next test run, from any lane) prunes it via `orphans()` before
103 * creating its own.
104 */
105export async function withEphemeralResource<T>(
106  provider: EphemeralResourceProvider<T>,
107  registry: ClaimRegistry,
108  fn: (resource: T) => Promise<void>,
109): Promise<void> {
110  for (const orphan of registry.orphans()) {
111    await provider.destroyOrphan(orphan);
112    registry.release(orphan.id);
113  }
114
115  const id = `${Date.now()}-${process.pid}`;
116  const claim: Claim = { id, pid: process.pid, createdAt: Date.now() };
117  registry.record(claim);
118
119  const resource = await provider.create();
120  try {
121    await fn(resource);
122  } finally {
123    await provider.destroy(resource);
124    registry.release(id);
125  }
126}
§ 05 · DETAIL

源码判断

这套锁的优势是简单、可审计、无需 daemon；代价是只能协调一台机器。src/lib/queue-lock.ts:74 把状态放在 os tmpdir，跨机器不会共享；但这正好匹配项目定位：本地 parallel Claude Code agents 的交通灯，而不是云端 merge queue。

Generated · 2026-07-07Source · funador/claude-code-merge-queue↑ 返回索引