diff --git a/packages/loopover-miner/lib/worktree-allocator.d.ts b/packages/loopover-miner/lib/worktree-allocator.d.ts index acfa3a8b8d..a0500d3136 100644 --- a/packages/loopover-miner/lib/worktree-allocator.d.ts +++ b/packages/loopover-miner/lib/worktree-allocator.d.ts @@ -5,6 +5,7 @@ export type WorktreeAllocation = { repoFullName: string | null; status: "free" | "active"; ownerPid: number | null; + ownerHost: string | null; allocatedAt: string | null; }; @@ -12,13 +13,17 @@ export type WorktreeAllocator = { dbPath: string; worktreeBaseDir: string; maxConcurrency: number; + maxLeaseMs: number; processPid: number; + hostId: string; acquire(attemptId: string, repoFullName: string): WorktreeAllocation; release(attemptId: string): WorktreeAllocation | null; listSlots(): WorktreeAllocation[]; close(): void; }; +export const DEFAULT_MAX_LEASE_MS: number; + export function resolveWorktreeAllocatorDbPath(env?: Record): string; export function resolveWorktreeBaseDir(env?: Record): string; @@ -29,7 +34,10 @@ export function openWorktreeAllocator(options?: { dbPath?: string; worktreeBaseDir?: string; maxConcurrency?: number; + maxLeaseMs?: number; processPid?: number; + hostId?: string; + nowMs?: number; }): WorktreeAllocator; export function acquireWorktree(attemptId: string, repoFullName: string): WorktreeAllocation; diff --git a/packages/loopover-miner/lib/worktree-allocator.js b/packages/loopover-miner/lib/worktree-allocator.js index dd660f60d1..865edc0800 100644 --- a/packages/loopover-miner/lib/worktree-allocator.js +++ b/packages/loopover-miner/lib/worktree-allocator.js @@ -2,7 +2,7 @@ // a filesystem directory, not a store DB path, and is deliberately out of this migration's scope. Only the DB // handle's own mkdir/chmod moved into openLocalStoreDb. import { mkdirSync } from "node:fs"; -import { homedir } from "node:os"; +import { homedir, hostname } from "node:os"; import { join } from "node:path"; import { normalizeLocalStoreDbPath, openLocalStoreDb, resolveLocalStoreDbPath } from "./local-store.js"; @@ -20,6 +20,17 @@ const defaultWorktreeDirName = "worktrees"; const defaultMaxConcurrency = 2; let defaultWorktreeAllocator = null; +// Age-based orphan reclaim (#7085). Fleet mode (see DEPLOYMENT.md) runs multiple separate CONTAINERS over one +// shared data volume, each with its own PID namespace, so a stored `owner_pid` is meaningless the moment a +// different container opens this store — `isProcessAlive` checks the CALLING process's own namespace, not the +// one that recorded the pid. So we mirror the age-based convention every sibling shared-lease store already uses +// (portfolio-queue-expiry.js's DEFAULT_MAX_LEASE_MS / sweepStuckItems, claim-ledger's DEFAULT_MAX_CLAIM_AGE_MS): +// reclaim any `active` slot older than this regardless of what the pid check reports. Kept well above +// portfolio-queue-expiry's 30-minute floor because a single worktree lease spans a whole coding attempt (clone + +// agent run + push), which can legitimately run for hours; the same-host `isProcessAlive` fast path still frees a +// crashed local owner immediately, so this age fallback only ever governs the cross-container case. +export const DEFAULT_MAX_LEASE_MS = 6 * 60 * 60 * 1000; + export function resolveWorktreeAllocatorDbPath(env = process.env) { return resolveLocalStoreDbPath(defaultDbFileName, "LOOPOVER_MINER_WORKTREE_ALLOCATOR_DB", env); } @@ -57,6 +68,18 @@ function normalizeMaxConcurrency(value) { return value; } +function normalizeMaxLeaseMs(value) { + if (value === undefined || value === null) return DEFAULT_MAX_LEASE_MS; + if (!Number.isFinite(value) || value < 0) throw new Error("invalid_max_lease_ms"); + return value; +} + +function normalizeHostId(value) { + if (value === undefined || value === null) return hostname(); + if (typeof value !== "string" || !value.trim()) throw new Error("invalid_host_id"); + return value.trim(); +} + function normalizeRepoFullName(repoFullName) { if (typeof repoFullName !== "string") throw new Error("invalid_repo_full_name"); const [owner, repo, extra] = repoFullName.trim().split("/"); @@ -92,6 +115,7 @@ function rowToAllocation(row) { repoFullName: row.repo_full_name, status: row.status, ownerPid: row.owner_pid, + ownerHost: row.owner_host ?? null, allocatedAt: row.allocated_at, }; } @@ -105,9 +129,23 @@ function ensureSlotTable(db) { repo_full_name TEXT, status TEXT NOT NULL CHECK (status IN ('free', 'active')), owner_pid INTEGER, + owner_host TEXT, allocated_at TEXT ) `); + ensureOwnerHostColumn(db); +} + +// Add the owner_host column (#7085) to an on-disk file created before it existed. `CREATE TABLE IF NOT EXISTS` +// above is a no-op against an already-existing table, so a pre-#7085 file needs this explicit ALTER — guarded by +// a presence check (same technique as attempt-log.js's ensureOutcomeColumns). A migrated row keeps owner_host +// NULL until its owner re-acquires, so the age-based reclaim (not the same-host pid fast path) governs it. +function ensureOwnerHostColumn(db) { + const hasOwnerHost = db + .prepare("PRAGMA table_info(worktree_slots)") + .all() + .some((column) => column.name === "owner_host"); + if (!hasOwnerHost) db.exec("ALTER TABLE worktree_slots ADD COLUMN owner_host TEXT"); } function ensureSlots(db, worktreeBaseDir, maxConcurrency) { @@ -123,41 +161,70 @@ function ensureSlots(db, worktreeBaseDir, maxConcurrency) { } } -function reclaimOrphanedAllocations(db) { +function allocationAgeMs(allocatedAt, nowMs) { + const allocatedMs = Date.parse(allocatedAt); + if (!Number.isFinite(allocatedMs)) return null; + return nowMs - allocatedMs; +} + +/** + * Decide whether an `active` slot is orphaned and should be reclaimed. Two independent signals: + * - Age (container-agnostic): a slot whose `allocated_at` is older than `maxLeaseMs` is reclaimed regardless of + * what `isProcessAlive` reports, guaranteeing eventual reclaim even when a cross-container caller observes the + * owner's pid in the wrong PID namespace. This is the only signal that is sound across fleet mode's separate + * containers, so it must never be gated behind the pid check. + * - Same-host pid liveness (fast path): only when the slot was leased by a process on THIS host (`owner_host` + * matches) is `isProcessAlive` a meaningful signal — a confirmed-dead (or missing) local owner frees its slot + * immediately without waiting out the lease. A foreign `owner_host` is never trusted for the pid check. + */ +function isSlotOrphaned(row, nowMs, maxLeaseMs, hostId) { + const ageMs = allocationAgeMs(row.allocated_at, nowMs); + if (ageMs !== null && ageMs > maxLeaseMs) return true; + if (row.owner_host !== null && row.owner_host === hostId) { + return row.owner_pid === null || !isProcessAlive(row.owner_pid); + } + return false; +} + +function reclaimOrphanedAllocations(db, nowMs, maxLeaseMs, hostId) { const orphans = db - .prepare("SELECT slot_index, owner_pid FROM worktree_slots WHERE status = 'active'") + .prepare("SELECT slot_index, owner_pid, owner_host, allocated_at FROM worktree_slots WHERE status = 'active'") .all(); const reclaim = db.prepare(` UPDATE worktree_slots - SET status = 'free', attempt_id = NULL, repo_full_name = NULL, owner_pid = NULL, allocated_at = NULL + SET status = 'free', attempt_id = NULL, repo_full_name = NULL, owner_pid = NULL, owner_host = NULL, allocated_at = NULL WHERE slot_index = ? `); for (const row of orphans) { - if (row.owner_pid !== null && isProcessAlive(row.owner_pid)) continue; - reclaim.run(row.slot_index); + if (isSlotOrphaned(row, nowMs, maxLeaseMs, hostId)) reclaim.run(row.slot_index); } } /** - * Opens the local worktree allocator store. Reclaims orphaned active slots from dead owner processes on startup. + * Opens the local worktree allocator store. On startup reclaims orphaned active slots — any slot past its + * `maxLeaseMs` age (the container-agnostic guarantee for fleet mode's shared store), plus, as a same-host fast + * path, any slot whose owner pid is confirmed dead in THIS host's PID namespace. */ export function openWorktreeAllocator(options = {}) { const resolvedPath = normalizeDbPath(options.dbPath); const worktreeBaseDir = normalizeWorktreeBaseDir(options.worktreeBaseDir); const maxConcurrency = normalizeMaxConcurrency(options.maxConcurrency); + const maxLeaseMs = normalizeMaxLeaseMs(options.maxLeaseMs); + const hostId = normalizeHostId(options.hostId); const processPid = Number.isInteger(options.processPid) ? options.processPid : process.pid; + const nowMs = Number.isFinite(options.nowMs) ? options.nowMs : Date.now(); const db = openLocalStoreDb(resolvedPath); ensureSlotTable(db); ensureSlots(db, worktreeBaseDir, maxConcurrency); - reclaimOrphanedAllocations(db); + reclaimOrphanedAllocations(db, nowMs, maxLeaseMs, hostId); const getByAttempt = db.prepare( - "SELECT slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, allocated_at FROM worktree_slots WHERE attempt_id = ?", + "SELECT slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, owner_host, allocated_at FROM worktree_slots WHERE attempt_id = ?", ); const countActive = db.prepare("SELECT COUNT(*) AS count FROM worktree_slots WHERE status = 'active'"); const selectFreeSlot = db.prepare(` - SELECT slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, allocated_at + SELECT slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, owner_host, allocated_at FROM worktree_slots WHERE status = 'free' ORDER BY slot_index @@ -165,24 +232,26 @@ export function openWorktreeAllocator(options = {}) { `); const markActive = db.prepare(` UPDATE worktree_slots - SET status = 'active', attempt_id = ?, repo_full_name = ?, owner_pid = ?, allocated_at = ? + SET status = 'active', attempt_id = ?, repo_full_name = ?, owner_pid = ?, owner_host = ?, allocated_at = ? WHERE slot_index = ? `); const releaseByAttempt = db.prepare(` UPDATE worktree_slots - SET status = 'free', attempt_id = NULL, repo_full_name = NULL, owner_pid = NULL, allocated_at = NULL + SET status = 'free', attempt_id = NULL, repo_full_name = NULL, owner_pid = NULL, owner_host = NULL, allocated_at = NULL WHERE attempt_id = ? AND status = 'active' - RETURNING slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, allocated_at + RETURNING slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, owner_host, allocated_at `); const listSlots = db.prepare( - "SELECT slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, allocated_at FROM worktree_slots ORDER BY slot_index", + "SELECT slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, owner_host, allocated_at FROM worktree_slots ORDER BY slot_index", ); const allocator = { dbPath: resolvedPath, worktreeBaseDir, maxConcurrency, + maxLeaseMs, processPid, + hostId, acquire(attemptId, repoFullName) { const normalizedAttempt = normalizeAttemptId(attemptId); const normalizedRepo = normalizeRepoFullName(repoFullName); @@ -201,7 +270,7 @@ export function openWorktreeAllocator(options = {}) { const slot = selectFreeSlot.get(); if (!slot) throw new Error("worktree_capacity_exceeded"); const allocatedAt = new Date().toISOString(); - markActive.run(normalizedAttempt, normalizedRepo, processPid, allocatedAt, slot.slot_index); + markActive.run(normalizedAttempt, normalizedRepo, processPid, hostId, allocatedAt, slot.slot_index); db.exec("COMMIT"); return rowToAllocation({ ...slot, @@ -209,6 +278,7 @@ export function openWorktreeAllocator(options = {}) { repo_full_name: normalizedRepo, status: "active", owner_pid: processPid, + owner_host: hostId, allocated_at: allocatedAt, }); } catch (error) { diff --git a/test/unit/miner-attempt-cli.test.ts b/test/unit/miner-attempt-cli.test.ts index 12628705fe..b082419996 100644 --- a/test/unit/miner-attempt-cli.test.ts +++ b/test/unit/miner-attempt-cli.test.ts @@ -1085,7 +1085,9 @@ describe("runAttempt (#5132)", () => { dbPath: ":memory:", worktreeBaseDir: "/tmp/unused", maxConcurrency: 1, + maxLeaseMs: 6 * 60 * 60 * 1000, processPid: process.pid, + hostId: "test-host", acquire: () => { throw new Error("no_free_worktree_slots"); }, diff --git a/test/unit/miner-worktree-allocator-lease-expiry.test.ts b/test/unit/miner-worktree-allocator-lease-expiry.test.ts new file mode 100644 index 0000000000..b67ed82dad --- /dev/null +++ b/test/unit/miner-worktree-allocator-lease-expiry.test.ts @@ -0,0 +1,226 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { afterEach, describe, expect, it } from "vitest"; +import { + DEFAULT_MAX_LEASE_MS, + openWorktreeAllocator, +} from "../../packages/loopover-miner/lib/worktree-allocator.js"; + +// Cross-container orphan reclaim (#7085): fleet mode runs separate CONTAINERS over one shared store, each with +// its own PID namespace, so `owner_pid` is a meaningless signal across containers. These tests seed active rows +// with controlled owner_pid/owner_host/allocated_at and open the allocator with an injected hostId + nowMs to +// prove the age-based reclaim (not the same-host pid fast path) governs the cross-container case. + +const roots: string[] = []; +const allocators: Array<{ close(): void }> = []; + +const NOW_MS = Date.parse("2026-07-17T12:00:00.000Z"); +const RECENT = "2026-07-17T11:59:00.000Z"; // ~1 minute before NOW_MS — well within any lease +const OLD = "2026-07-17T00:00:00.000Z"; // 12 hours before NOW_MS — past the 6h default lease +const DEAD_PID = 9_999_999; // absent in this test runner's PID namespace, so isProcessAlive() returns false + +function tempPaths() { + const root = mkdtempSync(join(tmpdir(), "loopover-miner-worktree-lease-")); + roots.push(root); + return { + dbPath: join(root, "worktree-allocator.sqlite3"), + worktreeBaseDir: join(root, "worktrees"), + }; +} + +/** Bootstrap the store schema/slots, then hand-write one active slot with fully controlled lease fields. */ +function seedActiveSlot( + paths: ReturnType, + seed: { ownerPid: number | null; ownerHost: string | null; allocatedAt: string | null }, +) { + const bootstrap = openWorktreeAllocator({ + dbPath: paths.dbPath, + worktreeBaseDir: paths.worktreeBaseDir, + maxConcurrency: 1, + }); + bootstrap.close(); + + const db = new DatabaseSync(paths.dbPath); + try { + db.prepare(` + UPDATE worktree_slots + SET status = 'active', + attempt_id = 'seeded-attempt', + repo_full_name = 'acme/widgets', + owner_pid = ?, + owner_host = ?, + allocated_at = ? + WHERE slot_index = 0 + `).run(seed.ownerPid, seed.ownerHost, seed.allocatedAt); + } finally { + db.close(); + } +} + +function reopen( + paths: ReturnType, + options: { hostId: string; nowMs?: number; maxLeaseMs?: number }, +) { + const allocator = openWorktreeAllocator({ + dbPath: paths.dbPath, + worktreeBaseDir: paths.worktreeBaseDir, + maxConcurrency: 1, + hostId: options.hostId, + nowMs: options.nowMs ?? NOW_MS, + ...(options.maxLeaseMs === undefined ? {} : { maxLeaseMs: options.maxLeaseMs }), + }); + allocators.push(allocator); + return allocator; +} + +function activeCount(allocator: ReturnType) { + return allocator.listSlots().filter((slot) => slot.status === "active").length; +} + +afterEach(() => { + for (const allocator of allocators.splice(0)) allocator.close(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +describe("loopover-miner worktree allocator age-based orphan reclaim (#7085)", () => { + it("exposes a lease default at least as generous as portfolio-queue-expiry's 30-minute floor", () => { + expect(DEFAULT_MAX_LEASE_MS).toBeGreaterThanOrEqual(30 * 60 * 1000); + }); + + it("does NOT reclaim a recent slot whose foreign owner_pid is dead in this namespace (cross-container false-negative)", () => { + // container-A recorded a still-running worker; container-B observes that pid as dead only because it is a + // different PID namespace. The age guard must protect the live worker's slot. + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: DEAD_PID, ownerHost: "container-A", allocatedAt: RECENT }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(1); + expect(() => allocator.acquire("intruder", "acme/other")).toThrow("worktree_capacity_exceeded"); + }); + + it("reclaims a stale slot whose foreign owner_pid collides with a live local pid (cross-container false-positive)", () => { + // container-A's recorded pid happens to match a live pid in container-B's namespace, so isProcessAlive would + // wrongly say "alive" — but the lease is past its deadline, so age reclaims it regardless. + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: process.pid, ownerHost: "container-A", allocatedAt: OLD }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(0); + expect(allocator.acquire("fresh", "acme/other").status).toBe("active"); + }); + + it("reclaims a past-lease slot even when its owner pid is alive on the SAME host (age overrides pid)", () => { + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: process.pid, ownerHost: "container-B", allocatedAt: OLD }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(0); + }); + + it("keeps a within-lease slot whose owner pid is alive on the SAME host", () => { + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: process.pid, ownerHost: "container-B", allocatedAt: RECENT }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(1); + }); + + it("reclaims a within-lease slot immediately when its owner pid is dead on the SAME host (fast path)", () => { + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: DEAD_PID, ownerHost: "container-B", allocatedAt: RECENT }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(0); + }); + + it("reclaims a within-lease slot with no recorded owner pid on the SAME host", () => { + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: null, ownerHost: "container-B", allocatedAt: RECENT }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(0); + }); + + it("keeps a within-lease legacy slot with no recorded owner_host until it ages out", () => { + // A row migrated from a pre-#7085 store has owner_host NULL, so the same-host pid fast path can never apply; + // only the age fallback governs it. + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: DEAD_PID, ownerHost: null, allocatedAt: RECENT }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(1); + }); + + it("reclaims a slot whose owner_host is unset once its lease is past due", () => { + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: DEAD_PID, ownerHost: null, allocatedAt: OLD }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(0); + }); + + it("reclaims a same-host slot with a dead owner even when allocated_at is missing", () => { + // Defensive: a corrupt active row with no parseable allocated_at yields a null age, so the age guard is + // skipped and the same-host pid check still frees it. + const paths = tempPaths(); + seedActiveSlot(paths, { ownerPid: DEAD_PID, ownerHost: "container-B", allocatedAt: null }); + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(0); + }); + + it("honors a caller-supplied maxLeaseMs shorter than the default", () => { + const paths = tempPaths(); + // 5 minutes old: within the 6h default, but past a 1-minute override. + seedActiveSlot(paths, { + ownerPid: process.pid, + ownerHost: "container-B", + allocatedAt: "2026-07-17T11:55:00.000Z", + }); + const allocator = reopen(paths, { hostId: "container-B", maxLeaseMs: 60_000 }); + expect(activeCount(allocator)).toBe(0); + }); + + it("rejects invalid lease and host configuration", () => { + const paths = tempPaths(); + const base = { dbPath: paths.dbPath, worktreeBaseDir: paths.worktreeBaseDir, maxConcurrency: 1 }; + expect(() => openWorktreeAllocator({ ...base, maxLeaseMs: -1 })).toThrow("invalid_max_lease_ms"); + expect(() => openWorktreeAllocator({ ...base, maxLeaseMs: Number.NaN })).toThrow("invalid_max_lease_ms"); + expect(() => openWorktreeAllocator({ ...base, hostId: " " })).toThrow("invalid_host_id"); + }); + + it("adds the owner_host column to a pre-#7085 store on open and reclaims its stale rows by age", () => { + const paths = tempPaths(); + // A file created before #7085 has no owner_host column. `CREATE TABLE IF NOT EXISTS` won't touch it, so the + // open path must ALTER it in — and the stale legacy active row (dead pid, past-due lease) must still reclaim. + const legacy = new DatabaseSync(paths.dbPath); + try { + legacy.exec(` + CREATE TABLE worktree_slots ( + slot_index INTEGER PRIMARY KEY, + worktree_path TEXT NOT NULL UNIQUE, + attempt_id TEXT UNIQUE, + repo_full_name TEXT, + status TEXT NOT NULL CHECK (status IN ('free', 'active')), + owner_pid INTEGER, + allocated_at TEXT + ) + `); + legacy + .prepare(` + INSERT INTO worktree_slots (slot_index, worktree_path, attempt_id, repo_full_name, status, owner_pid, allocated_at) + VALUES (0, ?, 'legacy-attempt', 'acme/widgets', 'active', ?, ?) + `) + .run(join(paths.worktreeBaseDir, "slot-0"), DEAD_PID, OLD); + } finally { + legacy.close(); + } + + const allocator = reopen(paths, { hostId: "container-B" }); + expect(activeCount(allocator)).toBe(0); + // A successful acquire proves the column was added — markActive writes owner_host and would throw otherwise. + expect(allocator.acquire("fresh", "acme/other").ownerHost).toBe("container-B"); + }); + + it("stamps the acquiring host and pid onto the allocation and clears them on release", () => { + const paths = tempPaths(); + const allocator = reopen(paths, { hostId: "container-B" }); + const allocation = allocator.acquire("attempt-a", "acme/widgets"); + expect(allocation.ownerHost).toBe("container-B"); + expect(allocation.ownerPid).toBe(process.pid); + expect(allocator.release("attempt-a")?.ownerHost).toBeNull(); + }); +});