Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 15 additions & 6 deletions packages/loopover-miner/lib/worktree-allocator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -239,17 +239,26 @@ function isSlotOrphaned(row: OrphanProbeRow, nowMs: number, maxLeaseMs: number,
return false;
}

function reclaimOrphanedAllocations(db: DatabaseSync, nowMs: number, maxLeaseMs: number, hostId: string): void {
const orphans = db
.prepare("SELECT slot_index, owner_pid, owner_host, allocated_at FROM worktree_slots WHERE status = 'active'")
.all() as OrphanProbeRow[];
/** Exported for the CAS regression test only — production callers go through the allocator handle. */
export function reclaimOrphanedAllocations(db: DatabaseSync, nowMs: number, maxLeaseMs: number, hostId: string, probedRows?: OrphanProbeRow[]): void {
const orphans =
probedRows ??
(db
.prepare("SELECT slot_index, owner_pid, owner_host, allocated_at FROM worktree_slots WHERE status = 'active'")
.all() as OrphanProbeRow[]);
// COMPARE-AND-SET on the exact lease evidence the probe saw: between the SELECT above and this UPDATE, a
// peer process can legitimately free-and-re-acquire the same slot (the sweep runs on EVERY acquire since
// #8859, so the window is hit under real concurrency — CI reproduced duplicate paths). A re-acquired slot
// carries a fresh allocated_at (and usually a new owner), so guarding on the probed values makes a stale
// reclaim a no-op instead of force-freeing a live peer's allocation and double-booking the worktree.
const reclaim = db.prepare(`
UPDATE worktree_slots
SET status = 'free', attempt_id = NULL, repo_full_name = NULL, owner_pid = NULL, owner_host = NULL, allocated_at = NULL
WHERE slot_index = ?
WHERE slot_index = ? AND status = 'active'
AND allocated_at IS ? AND owner_pid IS ? AND owner_host IS ?
`);
for (const row of orphans) {
if (isSlotOrphaned(row, nowMs, maxLeaseMs, hostId)) reclaim.run(row.slot_index);
if (isSlotOrphaned(row, nowMs, maxLeaseMs, hostId)) reclaim.run(row.slot_index, row.allocated_at, row.owner_pid, row.owner_host);
}
}

Expand Down
28 changes: 24 additions & 4 deletions test/fixtures/miner-worktree-allocator/acquire-child.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,13 @@
// Cross-process helper for worktree-allocator collision tests (#4298).
// Opens the shared store, waits for a stdin "go" signal, then calls acquire() so
// multiple Node processes contend on BEGIN IMMEDIATE against the same dbPath.
//
// LEASE LIFECYCLE (#8992): a child that acquires successfully HOLDS its lease (process alive, allocator
// open) until the test sends "done". Exiting right after acquire — the previous behavior — violated the
// lease contract the allocator is built around: the same-host dead-pid fast path in every LATER child's
// on-acquire sweep (#8859) then correctly reclaimed the dead child's slot and re-issued the same path,
// which is exactly what the distinct-paths and capacity tests flaked on under CI load (4-of-5, then
// 2-of-5 distinct). Production owners stay alive for the whole worktree lifetime; the fixture now does too.
import { openWorktreeAllocator } from "../../../packages/loopover-miner/dist/lib/worktree-allocator.js";

const [dbPath, worktreeBaseDir, maxConcurrencyStr, attemptId, repoFullName] = process.argv.slice(2);
Expand All @@ -17,23 +24,36 @@ const allocator = openWorktreeAllocator({
});

let started = false;
let holdingLease = false;

function runAcquire() {
if (started) return;
started = true;
try {
const allocation = allocator.acquire(attemptId, repoFullName);
holdingLease = true;
process.stdout.write(`${JSON.stringify({ ok: true, allocation })}\n`);
process.exit(0);
// Stay alive: the lease is held until the test says "done".
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
process.stdout.write(`${JSON.stringify({ ok: false, message })}\n`);
process.exit(1);
} finally {
allocator.close();
process.exit(1);
}
}

function finish() {
if (!holdingLease) return;
holdingLease = false;
allocator.close();
process.exit(0);
}

process.stdin.setEncoding("utf8");
process.stdin.on("data", () => runAcquire());
let stdinBuffer = "";
process.stdin.on("data", (chunk) => {
stdinBuffer += chunk;
if (stdinBuffer.includes("go\n")) runAcquire();
if (stdinBuffer.includes("done\n")) finish();
});
process.stdout.write("READY\n");
44 changes: 34 additions & 10 deletions test/unit/miner-worktree-allocator-collisions.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ function spawnAcquireChild(
attemptId: string,
maxConcurrency: number,
): ChildProcessWithoutNullStreams {
return spawn(
const child = spawn(
process.execPath,
[
acquireChildScript,
Expand All @@ -74,6 +74,10 @@ function spawnAcquireChild(
],
{ stdio: ["pipe", "pipe", "pipe"] },
);
// The done-broadcast can race a child that already exited (a rejected acquire exits immediately) — an
// unhandled EPIPE on its stdin must not crash the test run.
child.stdin.on("error", () => {});
return child;
}

async function waitForReady(child: ChildProcessWithoutNullStreams): Promise<void> {
Expand Down Expand Up @@ -102,29 +106,49 @@ async function runBarrieredAcquires(
const children = attemptIds.map((attemptId) => spawnAcquireChild(paths, attemptId, maxConcurrency));
await Promise.all(children.map((child) => waitForReady(child)));
for (const child of children) child.stdin.write("go\n");
return Promise.all(
// #8992: results resolve from STDOUT, not exit — a successful child HOLDS its lease (process alive) until
// every result is observed. The previous exit-after-acquire lifecycle let later children's on-acquire
// sweeps legitimately reclaim the dead winners' slots and re-issue the same paths (the CI 4-of-5 /
// 2-of-5-distinct flake), and could hand the capacity test a third success at a 2-slot cap.
const results = await Promise.all(
children.map(
(child) =>
new Promise<AcquireChildResult>((resolve, reject) => {
let stdout = "";
child.stdout.on("data", (chunk) => {
const onData = (chunk: Buffer | string) => {
stdout += chunk.toString();
});
child.once("error", reject);
child.once("exit", () => {
const line = stdout
.split("\n")
.map((entry) => entry.trim())
.find((entry) => entry.startsWith("{"));
if (!line) {
reject(new Error(`child produced no JSON result: ${stdout}`));
return;
if (line) {
child.stdout.off("data", onData);
resolve(JSON.parse(line) as AcquireChildResult);
}
resolve(JSON.parse(line) as AcquireChildResult);
};
child.stdout.on("data", onData);
child.once("error", reject);
child.once("exit", () => {
if (!stdout.includes("{")) reject(new Error(`child exited with no JSON result: ${stdout}`));
});
}),
),
);
// Every lease observed — release the survivors and wait for them to exit cleanly.
await Promise.all(
children.map(
(child) =>
new Promise<void>((resolve) => {
if (child.exitCode !== null) {
resolve();
return;
}
child.once("exit", () => resolve());
child.stdin.write("done\n");
}),
),
);
return results;
}

afterEach(() => {
Expand Down
29 changes: 29 additions & 0 deletions test/unit/miner-worktree-allocator.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
import { DatabaseSync } from "node:sqlite";
import {
acquireWorktree,
reclaimOrphanedAllocations,
closeDefaultWorktreeAllocator,
isProcessAlive,
openWorktreeAllocator,
Expand Down Expand Up @@ -231,3 +232,31 @@ describe("loopover-miner worktree allocator scaffolding (#4298)", () => {
});
});
});

describe("reclaim compare-and-set (#8918 follow-up: the duplicate-path race)", () => {
it("a stale probe snapshot must NOT free a slot a peer re-acquired between probe and apply", () => {
const dir = mkdtempSync(join(tmpdir(), "worktree-cas-"));
const dbPath = join(dir, "worktree-allocator.sqlite3");
const nowMs = Date.parse("2026-01-02T00:00:00.000Z");
const allocator = openWorktreeAllocator({ dbPath, worktreeBaseDir: join(dir, "wt"), maxConcurrency: 2, maxLeaseMs: 100, nowMs });
// An ancient lease: genuinely orphaned (a day past a 100ms lease) at probe time.
const db = new DatabaseSync(dbPath);
db.prepare("UPDATE worktree_slots SET status='active', attempt_id='ghost', owner_pid=999999, owner_host='other-host', allocated_at='2026-01-01T00:00:00.000Z' WHERE slot_index = 0").run();
const staleProbe = db.prepare("SELECT slot_index, owner_pid, owner_host, allocated_at FROM worktree_slots WHERE status = 'active'").all() as never;
// Between the probe and the apply, a PEER frees and re-acquires slot 0 — a fresh, live lease.
db.prepare("UPDATE worktree_slots SET attempt_id='live-peer', owner_pid=4242, owner_host='peer-host', allocated_at='2026-01-01T23:59:59.999Z' WHERE slot_index = 0").run();
// Applying the STALE snapshot must be a no-op: the lease evidence no longer matches.
reclaimOrphanedAllocations(db, nowMs, 100, "this-host", staleProbe);
const row = db.prepare("SELECT status, attempt_id FROM worktree_slots WHERE slot_index = 0").get() as { status: string; attempt_id: string };
expect(row.status).toBe("active");
expect(row.attempt_id).toBe("live-peer");
// And when the CURRENT row really is the aged-out lease (probe evidence matches), it frees.
db.prepare("UPDATE worktree_slots SET allocated_at='2026-01-01T00:00:00.000Z' WHERE slot_index = 0").run();
reclaimOrphanedAllocations(db, nowMs, 100, "this-host");
const freed = db.prepare("SELECT status FROM worktree_slots WHERE slot_index = 0").get() as { status: string };
expect(freed.status).toBe("free");
db.close();
allocator.close();
rmSync(dir, { recursive: true, force: true });
});
});