diff --git a/docs/mcp/DEPLOY.md b/docs/mcp/DEPLOY.md index 62bd84156..939084c52 100644 --- a/docs/mcp/DEPLOY.md +++ b/docs/mcp/DEPLOY.md @@ -258,6 +258,24 @@ the user owns the machine. See [ALTERNATIVES.md](ALTERNATIVES.md) for a comparison of ngrok, Tailscale Funnel, and cloud hosts (Fly.io, Railway). +### Running in a container (PID 1) + +When `gbrain serve` (or `gbrain jobs supervisor start`) is the container +entrypoint, it runs as PID 1 and inherits init duties: every orphaned +grandchild process the kernel re-parents onto it must be reaped, or zombies +accumulate until the container's pid limit is exhausted. Prefer a real init +as the entrypoint: + +```bash +docker run --init ... # Docker's bundled tini +# or in a Dockerfile: +ENTRYPOINT ["tini", "--", "gbrain", "serve", "--http"] +``` + +gbrain also ships an in-process backstop: when it detects it is PID 1 on +Linux it periodically reaps orphaned zombie children. An explicit init like +tini is still the recommended setup. + ## Troubleshooting **"missing_auth" error** diff --git a/llms-full.txt b/llms-full.txt index 4de2b11e4..7c57aa9bb 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -3902,6 +3902,24 @@ the user owns the machine. See [ALTERNATIVES.md](ALTERNATIVES.md) for a comparison of ngrok, Tailscale Funnel, and cloud hosts (Fly.io, Railway). +### Running in a container (PID 1) + +When `gbrain serve` (or `gbrain jobs supervisor start`) is the container +entrypoint, it runs as PID 1 and inherits init duties: every orphaned +grandchild process the kernel re-parents onto it must be reaped, or zombies +accumulate until the container's pid limit is exhausted. Prefer a real init +as the entrypoint: + +```bash +docker run --init ... # Docker's bundled tini +# or in a Dockerfile: +ENTRYPOINT ["tini", "--", "gbrain", "serve", "--http"] +``` + +gbrain also ships an in-process backstop: when it detects it is PID 1 on +Linux it periodically reaps orphaned zombie children. An explicit init like +tini is still the recommended setup. + ## Troubleshooting **"missing_auth" error** diff --git a/src/cli.ts b/src/cli.ts index 804d7dbf3..2e5ec749c 100755 --- a/src/cli.ts +++ b/src/cli.ts @@ -1,7 +1,13 @@ #!/usr/bin/env bun -import { installSigchldHandler } from './core/zombie-reap.ts'; +import { installSigchldHandler, installPid1OrphanReaper } from './core/zombie-reap.ts'; installSigchldHandler(); +// #2443: when gbrain IS the container init (PID 1 — `gbrain serve` / +// `gbrain jobs supervisor start` as the entrypoint), re-parented orphan +// grandchildren (e.g. git spawned by a dead worker) become zombies nobody +// waits on and eventually exhaust the cgroup pid budget. No-op unless +// PID 1 on Linux. +installPid1OrphanReaper(); // v0.41.6.0 D5: cleanup registry + signal handlers for SIGTERM/SIGHUP/SIGPIPE/ // uncaughtException. NOT SIGINT (the existing AbortController path at :254 // owns SIGINT). Installed at module load so locks acquired during boot diff --git a/src/core/facts/fence-write.ts b/src/core/facts/fence-write.ts index ae35769fb..cf283e670 100644 --- a/src/core/facts/fence-write.ts +++ b/src/core/facts/fence-write.ts @@ -33,11 +33,12 @@ * sees the constraint. */ -import { existsSync, mkdirSync, readFileSync, writeFileSync, renameSync, appendFileSync } from 'node:fs'; -import { dirname } from 'node:path'; +import { existsSync, mkdirSync, readFileSync, writeFileSync, renameSync, appendFileSync, unlinkSync } from 'node:fs'; +import { dirname, join } from 'node:path'; import type { BrainEngine, NewFact, FactVisibility } from '../engine.ts'; import { resolvePageFilePath } from '../markdown.ts'; +import { isWriteTargetContained } from '../path-confine.ts'; import { withPageLock } from '../page-lock.ts'; import { gbrainPath } from '../config.ts'; import { upsertFactRow, parseFactsFence } from '../facts-fence.ts'; @@ -92,6 +93,38 @@ export interface FenceWriteResult { const FAILURE_LOG_PATH = (): string => gbrainPath('facts.write_failures.jsonl'); +/** + * #2722: compute the on-disk file to fence into. When the page row exists + * with an import-time `pages.source_path`, that path (relative to the + * source's local_path — the same root `scanOneSource` walked) is the + * canonical file, and it can differ from the slug-derived name (slugs are + * normalized: `10 People/Jane Doe.md` → slug `10-people/jane-doe`). Only + * honored for markdown files that stay inside the source root (the DB row + * is data, not trusted path input). Everything else falls back to the + * slug-derived path, which is also the stub-create location for genuinely + * new pages. + * + * Shared with `forgetFactInFence` so the forget path edits the SAME file + * the fence write deposited into. + */ +export async function resolveFenceFilePath(engine: BrainEngine, target: FenceTarget): Promise { + const root = target.localPath as string; + try { + const rows = await engine.executeRaw<{ source_path: string | null }>( + `SELECT source_path FROM pages WHERE source_id = $1 AND slug = $2 LIMIT 1`, + [target.sourceId, target.slug], + ); + const sourcePath = rows[0]?.source_path ?? null; + if (sourcePath && sourcePath.toLowerCase().endsWith('.md')) { + const candidate = join(root, sourcePath); + if (isWriteTargetContained(candidate, root)) return candidate; + } + } catch { + // Best-effort lookup — fall back to the slug-derived path. + } + return resolvePageFilePath(root, target.slug, target.sourceId); +} + function recordWriteFailure(slug: string, sourceId: string, warnings: string[], filePath: string): void { // Best-effort JSONL append — never throws back into the caller. The // log is the operator-visibility surface; `gbrain doctor` reads it @@ -172,7 +205,14 @@ export async function writeFactsToFence( // the put_page write-through and dream-cycle reverse-render compute. The // bare join wrote main-source fences to the repo ROOT (the default source's // tree), polluting ~/brain with stray root-level fence files. - const filePath = resolvePageFilePath(target.localPath, target.slug, target.sourceId); + // + // #2722: when the page row already exists with an import-time + // `pages.source_path`, fence into THAT file. Slugs are normalized (spaces → + // hyphens, lowercase), so a page imported from `10 People/Jane Doe.md` has + // slug `10-people/jane-doe` — the slug-derived path would stub-create a + // parallel `10-people/jane-doe.md` next to the canonical page and fork it. + // The slug-derived path stays the fallback for genuinely new pages. + const filePath = await resolveFenceFilePath(engine, target); const tmpPath = `${filePath}.tmp`; return withPageLock( @@ -180,6 +220,7 @@ export async function writeFactsToFence( async () => { // 1. Read existing body or stub-create. let body: string; + let stubCreated = false; if (existsSync(filePath)) { body = readFileSync(filePath, 'utf-8'); } else { @@ -215,6 +256,7 @@ export async function writeFactsToFence( // Stub-create the parent directory if it doesn't exist. mkdirSync(dirname(filePath), { recursive: true }); body = stubEntityPage(target.slug); + stubCreated = true; } // 2. Upsert each fact onto the fence in input order. row_num @@ -274,7 +316,20 @@ export async function writeFactsToFence( source_session: facts[i].sessionId, })); - const result = await engine.insertFacts(enriched, { source_id: target.sourceId }); // gbrain-allow-direct-insert: writeFactsToFence is the markdown-first reconcile path; runs only after the atomic fence write commits + let result: { inserted: number; ids: number[] }; + try { + result = await engine.insertFacts(enriched, { source_id: target.sourceId }); // gbrain-allow-direct-insert: writeFactsToFence is the markdown-first reconcile path; runs only after the atomic fence write commits + } catch (err) { + // #2722 (secondary): if the DB stamp fails on a page we JUST + // stub-created, remove the stub so a transient DB error doesn't + // leave a phantom entity page on disk with facts the index never + // recorded. Pre-existing pages keep their fence (system of record; + // doctor/extract reconciles the index later). + if (stubCreated) { + try { unlinkSync(filePath); } catch { /* best effort */ } + } + throw err; + } return { inserted: result.inserted, ids: result.ids }; }, { timeoutMs: 5_000 }, diff --git a/src/core/facts/forget.ts b/src/core/facts/forget.ts index d667874de..22425e016 100644 --- a/src/core/facts/forget.ts +++ b/src/core/facts/forget.ts @@ -33,11 +33,11 @@ */ import { existsSync, readFileSync, writeFileSync, renameSync } from 'node:fs'; -import { join } from 'node:path'; import type { BrainEngine } from '../engine.ts'; import { withPageLock } from '../page-lock.ts'; import { parseFactsFence, renderFactsTable, type ParsedFact } from '../facts-fence.ts'; +import { resolveFenceFilePath } from './fence-write.ts'; export interface ForgetFactResult { /** True iff the row was found AND a forget was applied (fence or DB). */ @@ -126,7 +126,12 @@ export async function forgetFactInFence( const slug = row.source_markdown_slug!; const targetRowNum = row.row_num!; - const filePath = join(localPath, `${slug}.md`); + // #2722: resolve through the same helper the fence WRITE used, so a page + // imported from a spaced filename (slug `10-people/jane-doe`, file + // `10 People/Jane Doe.md`) gets its forget struck through in the canonical + // file — a bare `join(localPath, slug + '.md')` missed it and silently + // degraded to the DB-only path (forget didn't survive rebuild). + const filePath = await resolveFenceFilePath(engine, { sourceId: row.source_id, localPath, slug }); const tmpPath = `${filePath}.tmp`; if (!existsSync(filePath)) { diff --git a/src/core/minions/supervisor.ts b/src/core/minions/supervisor.ts index 9a71355d1..d072c496b 100644 --- a/src/core/minions/supervisor.ts +++ b/src/core/minions/supervisor.ts @@ -44,11 +44,12 @@ import { } from 'fs'; import { dirname } from 'path'; import type { BrainEngine } from '../engine.ts'; -import { tryAcquireDbLock, type DbLockHandle } from '../db-lock.ts'; +import { tryAcquireDbLock, inspectLock, classifyHolderLiveness, resolveStealGraceSeconds, type DbLockHandle, type LockSnapshot } from '../db-lock.ts'; import { currentBrainId } from './worker-registry.ts'; export type SupervisorEvent = | 'started' + | 'db_lock_wait' | 'worker_spawned' | 'worker_exited' | 'worker_spawn_failed' @@ -334,6 +335,96 @@ export function supervisorLockId(queue: string): string { return `gbrain-supervisor:${queue}`; } +/** + * #2308: acquire the supervisor DB lock, waiting out a DEAD holder instead of + * one-shot exiting. + * + * Container recreation is the motivating case: the old container's supervisor + * dies without releasing the row, the replacement container has a new hostname + * (so same-host dead-PID takeover can't fire), and the row only becomes + * stealable once its TTL + heartbeat steal-grace lapse (~5-6 min). A one-shot + * acquire at boot exits LOCK_HELD inside that window and nothing retries — the + * queue is dead until an operator intervenes. + * + * A single snapshot cannot distinguish "live supervisor elsewhere" from + * "holder died seconds ago" — both look like an unexpired, recently-refreshed + * row. Liveness is therefore decided by OBSERVATION: + * - Same-host holder whose PID is alive → return null immediately (fail + * fast; the interactive duplicate-start case keeps its instant exit 2). + * - Heartbeat ADVANCES between polls → a live supervisor is refreshing + * somewhere (e.g. another container) → return null. + * - Heartbeat never advances → the holder is dead; keep polling until the + * TTL + steal grace lapse and the upsert's host-agnostic takeover branch + * fires. Bounded by maxWaitMs (default TTL + steal grace + 30s margin). + * + * Exported for tests; the seams replace the DB round-trips and the sleep. + */ +export interface SupervisorLockWaitOpts { + maxWaitMs?: number; + retryMs?: number; + tryAcquire?: () => Promise; + inspect?: () => Promise; + sleep?: (ms: number) => Promise; + onWait?: (snap: LockSnapshot, waitedMs: number) => void; + now?: () => number; + /** Test seam for the same-host PID probe (defaults to process.kill via classifyHolderLiveness). */ + processKill?: (pid: number, signal: number) => void; +} + +export async function acquireSupervisorDbLockWithWait( + engine: BrainEngine, + lockId: string, + ttlMinutes: number, + opts: SupervisorLockWaitOpts = {}, +): Promise { + const tryAcquire = opts.tryAcquire ?? (() => tryAcquireDbLock(engine, lockId, ttlMinutes)); + const inspect = opts.inspect ?? (() => inspectLock(engine, lockId)); + const sleep = opts.sleep ?? ((ms: number) => new Promise((r) => setTimeout(r, ms))); + const now = opts.now ?? Date.now; + const retryMs = opts.retryMs ?? 10_000; + const maxWaitMs = opts.maxWaitMs + ?? ttlMinutes * 60_000 + resolveStealGraceSeconds(ttlMinutes) * 1000 + 30_000; + + const first = await tryAcquire(); + if (first) return first; + + const start = now(); + let lastSeenRefresh: number | null = null; + for (;;) { + let snap: LockSnapshot | null = null; + try { + snap = await inspect(); + } catch { + // Transient inspect failure — treat as unknown and keep polling. + } + if (snap) { + // Same-host live PID → a supervisor is genuinely running here. + const liveness = classifyHolderLiveness( + snap.holder_pid, + snap.holder_host, + snap.age_ms, + opts.processKill ? { processKill: opts.processKill } : {}, + ); + if (liveness === 'alive') return null; + // Heartbeat advanced since the last poll → a live holder is refreshing + // (cross-host, unprobeable) → fail fast. + const refreshedAt = snap.last_refreshed_at?.getTime() ?? null; + if (lastSeenRefresh !== null && refreshedAt !== null && refreshedAt > lastSeenRefresh) { + return null; + } + if (refreshedAt !== null) lastSeenRefresh = refreshedAt; + } + + const waited = now() - start; + if (waited >= maxWaitMs) return null; + if (snap) opts.onWait?.(snap, waited); + + await sleep(retryMs); + const handle = await tryAcquire(); + if (handle) return handle; + } +} + /** * #1849 (doctor): pure classification of the supervisor singleton state from * the DB lock holder vs the local pidfile holder. Compares host+pid (bare pid @@ -545,7 +636,26 @@ export class MinionSupervisor { // above but loses here, so it can't run a conflicting --max-rss worker on // the same (db, queue). Keyed on the queue alone; the database half of the // mutex is physical (the lock row lives in this DB). - this.dbLock = await tryAcquireDbLock(this.engine, this.supervisorLockId(), SUPERVISOR_LOCK_TTL_MIN); + // #2308: wait out a DEAD holder (e.g. the previous container's supervisor + // whose row can only lapse via TTL, since cross-host takeover is TTL-only) + // instead of one-shot exiting inside the TTL window. A genuinely live + // holder (same-host live PID, or a heartbeat that advances) still exits + // LOCK_HELD. + this.dbLock = await acquireSupervisorDbLockWithWait( + this.engine, + this.supervisorLockId(), + SUPERVISOR_LOCK_TTL_MIN, + { + onWait: (snap, waitedMs) => { + this.emit('db_lock_wait', { + holder_pid: snap.holder_pid, + holder_host: snap.holder_host, + ms_since_last_refresh: snap.ms_since_last_refresh, + waited_ms: waitedMs, + }); + }, + }, + ); if (!this.dbLock) { console.error( `Supervisor already running for queue '${this.opts.queue}' on this database ` + diff --git a/src/core/write-through.ts b/src/core/write-through.ts index 02f792a3f..d71a1a6e9 100644 --- a/src/core/write-through.ts +++ b/src/core/write-through.ts @@ -21,13 +21,21 @@ * only does "row exists + repo is a real dir → render + atomic write". */ -import { existsSync, statSync, mkdirSync, writeFileSync, renameSync, unlinkSync } from 'fs'; +import { existsSync, statSync, mkdirSync, readFileSync, writeFileSync, renameSync, unlinkSync } from 'fs'; import { dirname, join } from 'path'; import { randomBytes } from 'crypto'; import type { BrainEngine } from './engine.ts'; import { serializePageToMarkdown, resolvePageFilePath } from './markdown.ts'; import { isWriteTargetContained } from './path-confine.ts'; import { isDurabilityHardened, commitWriteThroughFile } from './brain-repo-durability.ts'; +import { withPageLock } from './page-lock.ts'; +import { + parseFactsFence, + renderFactsTable, + FACTS_FENCE_BEGIN, + FACTS_FENCE_END, + type ParsedFact, +} from './facts-fence.ts'; /** Minimal logger surface — structurally compatible with operations.ts `Logger`. */ export interface WriteThroughLogger { @@ -62,6 +70,38 @@ export interface WriteThroughResult { error?: string; } +/** + * #2839: fence writers (`writeFactsToFence`, `forgetFactInFence`) edit the + * on-disk `## Facts` fence FIRST and stamp the DB facts index second — the + * `pages.body` row only catches up on the next sync/import. A write-through + * render from the (lagging) DB row would silently drop those just-deposited + * fence rows from disk. Merge them back: union by row_num, the DISK row wins + * a collision (the fence file is the system of record for facts; any fence + * state inside the DB body is at best the last-imported snapshot). + * + * Exported for tests. + */ +export function mergeDiskFactsFence(renderedMd: string, diskBody: string): string { + const disk = parseFactsFence(diskBody); + if (disk.facts.length === 0) return renderedMd; + + const rendered = parseFactsFence(renderedMd); + const byRowNum = new Map(); + for (const f of rendered.facts) byRowNum.set(f.rowNum, f); + for (const f of disk.facts) byRowNum.set(f.rowNum, f); // disk wins + const merged = [...byRowNum.values()].sort((a, b) => a.rowNum - b.rowNum); + const fence = renderFactsTable(merged); + + const beginIdx = renderedMd.indexOf(FACTS_FENCE_BEGIN); + const endIdx = renderedMd.indexOf(FACTS_FENCE_END, beginIdx + FACTS_FENCE_BEGIN.length); + if (beginIdx !== -1 && endIdx !== -1 && endIdx > beginIdx) { + return renderedMd.slice(0, beginIdx) + fence + renderedMd.slice(endIdx + FACTS_FENCE_END.length); + } + // Rendered body has no fence at all — append one (same shape upsertFactRow creates). + const sep = renderedMd.endsWith('\n') ? '\n' : '\n\n'; + return `${renderedMd}${sep}## Facts\n\n${fence}\n`; +} + export interface WritePageThroughOpts { sourceId?: string; /** Merged over the page's own frontmatter at render time (e.g. provenance). */ @@ -148,22 +188,40 @@ export async function writePageThrough( mkdirSync(dirname(filePath), { recursive: true }); - // Atomic write: unique temp sibling + rename. Unique name (pid + random) - // so two concurrent saves to the same target can't clobber each other's - // temp file. Clean up the temp on any failure so we never leak a stray - // `.tmp` next to the real file. - const tmpPath = `${filePath}.tmp.${process.pid}.${randomBytes(4).toString('hex')}`; - try { - writeFileSync(tmpPath, md, 'utf8'); - renameSync(tmpPath, filePath); - } catch (writeErr) { + // #2839: serialize against the fence writers (writeFactsToFence / + // forgetFactInFence hold the same per-slug lock) so this read-merge- + // rename can't interleave with a fence edit, and merge any on-disk + // fence rows the DB body hasn't caught up with before rendering over + // the file. Lock timeout surfaces via the outer catch → `error` field + // (writePageThrough never throws). + await withPageLock(slug, async () => { + let out = md; try { - if (existsSync(tmpPath)) unlinkSync(tmpPath); + if (existsSync(filePath)) { + out = mergeDiskFactsFence(md, readFileSync(filePath, 'utf8')); + } } catch { - // best-effort cleanup; surface the original write error below + // Unreadable existing file — fall back to the rendered body. + out = md; } - throw writeErr; - } + + // Atomic write: unique temp sibling + rename. Unique name (pid + random) + // so two concurrent saves to the same target can't clobber each other's + // temp file. Clean up the temp on any failure so we never leak a stray + // `.tmp` next to the real file. + const tmpPath = `${filePath}.tmp.${process.pid}.${randomBytes(4).toString('hex')}`; + try { + writeFileSync(tmpPath, out, 'utf8'); + renameSync(tmpPath, filePath); + } catch (writeErr) { + try { + if (existsSync(tmpPath)) unlinkSync(tmpPath); + } catch { + // best-effort cleanup; surface the original write error below + } + throw writeErr; + } + }, { timeoutMs: 5_000 }); // #2426: on a durability-hardened repo (user ran `gbrain sources harden`), // commit the artifact so it reaches git — pre-fix, write-through content diff --git a/src/core/zombie-reap.ts b/src/core/zombie-reap.ts index 4946e3e08..773cfb087 100644 --- a/src/core/zombie-reap.ts +++ b/src/core/zombie-reap.ts @@ -14,6 +14,8 @@ * same process). */ +import { readdirSync, readFileSync } from 'node:fs'; + const reapHandler = () => {}; export function installSigchldHandler(): void { @@ -34,3 +36,158 @@ export function installSigchldHandler(): void { export function _uninstallSigchldHandlerForTests(): void { process.removeListener('SIGCHLD', reapHandler); } + +/** + * #2443: PID-1 orphan reaper. + * + * The SIGCHLD handler above only makes the runtime reap its OWN children. + * When gbrain runs as PID 1 (a container entrypoint: `gbrain serve`, + * `gbrain jobs supervisor start`), the kernel re-parents every orphaned + * grandchild (e.g. `git` spawned by a worker that died) onto us — and + * nothing waits on them. The zombies accumulate until the cgroup pids.max + * is exhausted and fork starts failing with EAGAIN. + * + * We can't blanket `waitpid(-1)` on SIGCHLD: that races the runtime's own + * per-child reaping and can steal exit statuses from Bun.spawn / + * child_process (exit events never fire → hangs). Instead: a periodic + * /proc sweep collects Z-state children of PID 1 and only reaps a pid seen + * in TWO consecutive sweeps. The runtime reaps its own children within + * milliseconds of exit, so anything still a zombie a full sweep interval + * later is an orphan (or native-spawned) that only we can collect. + * waitpid targets the specific pid with WNOHANG, so a mistaken candidate + * costs nothing. + * + * Linux-only by construction (/proc + the container PID-1 scenario). + * Running under an init like tini/`docker run --init` remains the + * recommended deployment (see docs/mcp/DEPLOY.md); this is the in-process + * backstop for images that don't. + */ + +/** Parse one /proc//stat line → { pid, state, ppid }, or null. */ +export function parseProcStatLine(stat: string): { pid: number; state: string; ppid: number } | null { + // Format: `pid (comm) state ppid ...` — comm may contain spaces and + // parens, so split on the LAST ')'. + const close = stat.lastIndexOf(')'); + if (close === -1) return null; + const pid = parseInt(stat.slice(0, stat.indexOf(' ')), 10); + const rest = stat.slice(close + 1).trim().split(/\s+/); + if (!Number.isFinite(pid) || rest.length < 2) return null; + const ppid = parseInt(rest[1], 10); + if (!Number.isFinite(ppid)) return null; + return { pid, state: rest[0], ppid }; +} + +/** List Z-state children of this process by walking /proc. Linux only. */ +function listZombieChildrenViaProc(selfPid: number): number[] { + const out: number[] = []; + let entries: string[]; + try { + entries = readdirSync('/proc'); + } catch { + return out; + } + for (const entry of entries) { + if (!/^\d+$/.test(entry)) continue; + try { + const parsed = parseProcStatLine(readFileSync(`/proc/${entry}/stat`, 'utf8')); + if (parsed && parsed.state === 'Z' && parsed.ppid === selfPid) out.push(parsed.pid); + } catch { + // Process vanished mid-walk — fine. + } + } + return out; +} + +/** Lazily-loaded libc waitpid via bun:ffi. null = unavailable (not Bun / no libc match). */ +let cachedWaitpid: ((pid: number) => void) | null | undefined; +async function loadWaitpid(): Promise<((pid: number) => void) | null> { + if (cachedWaitpid !== undefined) return cachedWaitpid; + cachedWaitpid = null; + try { + const { dlopen, FFIType, ptr } = await import('bun:ffi'); + const candidates = process.platform === 'darwin' + ? ['libSystem.dylib'] + : ['libc.so.6', 'libc.so', 'libc.musl-x86_64.so.1', 'libc.musl-aarch64.so.1']; + for (const name of candidates) { + try { + const lib = dlopen(name, { + waitpid: { args: [FFIType.i32, FFIType.ptr, FFIType.i32], returns: FFIType.i32 }, + }); + const status = new Int32Array(1); + const WNOHANG = 1; + cachedWaitpid = (pid: number) => { lib.symbols.waitpid(pid, ptr(status), WNOHANG); }; + break; + } catch { + // try next candidate + } + } + } catch { + // Not running under Bun — no FFI, reaper stays inert. + } + return cachedWaitpid; +} + +export interface OrphanReaperSweeper { + /** Run one sweep. Exposed so tests can drive it deterministically. */ + sweep: () => Promise; +} + +/** + * Two-sweep-confirmation sweeper core (exported for tests; seams replace + * the /proc walk and the FFI waitpid). + */ +export function createOrphanSweeper( + listZombieChildren: () => number[], + reapPid: (pid: number) => void, +): OrphanReaperSweeper { + let pendingFromLastSweep = new Set(); + return { + sweep: async () => { + const zombies = listZombieChildren(); + const next = new Set(); + for (const pid of zombies) { + if (pendingFromLastSweep.has(pid)) { + try { reapPid(pid); } catch { /* best effort */ } + } else { + next.add(pid); + } + } + pendingFromLastSweep = next; + }, + }; +} + +export interface InstallOrphanReaperOpts { + /** Test seams. */ + pid?: number; + platform?: NodeJS.Platform; + intervalMs?: number; + listZombieChildren?: () => number[]; + reapPid?: (pid: number) => void; +} + +const ORPHAN_SWEEP_INTERVAL_MS = 30_000; + +/** + * Install the PID-1 orphan reaper. No-op (returns null) unless this process + * is PID 1 on Linux. Returns a stop function otherwise. The interval is + * unref'd so it never keeps the process alive. + */ +export function installPid1OrphanReaper(opts: InstallOrphanReaperOpts = {}): (() => void) | null { + const pid = opts.pid ?? process.pid; + const platform = opts.platform ?? process.platform; + if (pid !== 1 || platform !== 'linux') return null; + + const list = opts.listZombieChildren ?? (() => listZombieChildrenViaProc(pid)); + const sweeper = createOrphanSweeper(list, (zombiePid) => { + if (opts.reapPid) { + opts.reapPid(zombiePid); + return; + } + void loadWaitpid().then((waitpid) => { waitpid?.(zombiePid); }); + }); + + const timer = setInterval(() => { void sweeper.sweep(); }, opts.intervalMs ?? ORPHAN_SWEEP_INTERVAL_MS); + (timer as { unref?: () => void }).unref?.(); + return () => clearInterval(timer); +} diff --git a/test/fence-write.test.ts b/test/fence-write.test.ts index 3bcb0c05b..9ed72f4b9 100644 --- a/test/fence-write.test.ts +++ b/test/fence-write.test.ts @@ -16,7 +16,7 @@ import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; import { mkdtempSync, rmSync, existsSync, readFileSync, writeFileSync, mkdirSync } from 'node:fs'; import { tmpdir } from 'node:os'; -import { join } from 'node:path'; +import { join, dirname } from 'node:path'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; import { writeFactsToFence, lookupSourceLocalPath } from '../src/core/facts/fence-write.ts'; @@ -292,6 +292,105 @@ describe('lookupSourceLocalPath', () => { }); }); +describe('writeFactsToFence — pages.source_path routing (#2722)', () => { + test('fences into the import-time source_path file, not a slug-derived sibling', async () => { + // Canonical page imported from a spaced filename: slug normalization + // means the slug-derived path differs from the real file. + const { importFromContent } = await import('../src/core/import-file.ts'); + const canonicalRel = '10 People/Jane Doe.md'; + const canonicalAbs = join(brainDir, canonicalRel); + mkdirSync(dirname(canonicalAbs), { recursive: true }); + const content = '---\ntype: person\ntitle: Jane Doe\n---\n\n# Jane Doe\n\nMet at demo day.\n'; + writeFileSync(canonicalAbs, content, 'utf-8'); + const imported = await importFromContent(engine, '10-people/jane-doe', content, { + noEmbed: true, + sourceId: 'default', + sourcePath: canonicalRel, + }); + expect(imported).toBeTruthy(); + + const result = await writeFactsToFence( + engine, + { sourceId: 'default', localPath: brainDir, slug: '10-people/jane-doe' }, + [baseInput({ fact: 'Raised a seed round' })], + ); + + expect(result.inserted).toBe(1); + // Fence landed in the canonical spaced-name file... + const body = readFileSync(canonicalAbs, 'utf-8'); + expect(body).toContain('Met at demo day.'); + expect(body).toContain('## Facts'); + expect(body).toContain('Raised a seed round'); + // ...and no hyphen-path fork was stub-created next to it. + expect(existsSync(join(brainDir, '10-people/jane-doe.md'))).toBe(false); + }); + + test('falls back to the slug-derived path when no page row exists', async () => { + const result = await writeFactsToFence( + engine, + { sourceId: 'default', localPath: brainDir, slug: 'people/newcomer' }, + [baseInput()], + ); + expect(result.inserted).toBe(1); + expect(existsSync(join(brainDir, 'people/newcomer.md'))).toBe(true); + }); + + test('ignores a hostile source_path that escapes the source root', async () => { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + await (engine as any).db.query( + `INSERT INTO pages (slug, title, type, source_id, source_path) + VALUES ('people/evil', 'Evil', 'person', 'default', '../../escape.md')`, + ); + const result = await writeFactsToFence( + engine, + { sourceId: 'default', localPath: brainDir, slug: 'people/evil' }, + [baseInput()], + ); + expect(result.inserted).toBe(1); + // Fell back to the slug-derived, contained path. + expect(existsSync(join(brainDir, 'people/evil.md'))).toBe(true); + expect(existsSync(join(brainDir, '..', '..', 'escape.md'))).toBe(false); + }); +}); + +describe('writeFactsToFence — stub cleanup on DB failure (#2722)', () => { + const failingEngine = () => ({ + kind: engine.kind, + executeRaw: engine.executeRaw.bind(engine), + insertFacts: async () => { + throw new Error('db down'); + }, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + }) as any; + + test('removes a freshly stub-created page when the DB stamp fails', async () => { + await expect( + writeFactsToFence( + failingEngine(), + { sourceId: 'default', localPath: brainDir, slug: 'people/ghost' }, + [baseInput()], + ), + ).rejects.toThrow('db down'); + expect(existsSync(join(brainDir, 'people/ghost.md'))).toBe(false); + }); + + test('keeps a pre-existing page file when the DB stamp fails (fence stays system of record)', async () => { + const filePath = join(brainDir, 'people/survivor.md'); + mkdirSync(dirname(filePath), { recursive: true }); + writeFileSync(filePath, '---\ntype: person\ntitle: S\nslug: people/survivor\n---\n\n# S\n', 'utf-8'); + await expect( + writeFactsToFence( + failingEngine(), + { sourceId: 'default', localPath: brainDir, slug: 'people/survivor' }, + [baseInput()], + ), + ).rejects.toThrow('db down'); + expect(existsSync(filePath)).toBe(true); + // The fence write itself committed (system of record); only the DB stamp failed. + expect(readFileSync(filePath, 'utf-8')).toContain('## Facts'); + }); +}); + // Cleanup any leftover tempdirs after the whole suite. afterAll(() => { // No-op: each test cleaned up via the beforeEach; this is a safety net. diff --git a/test/supervisor-db-lock-wait.test.ts b/test/supervisor-db-lock-wait.test.ts new file mode 100644 index 000000000..ae58e15df --- /dev/null +++ b/test/supervisor-db-lock-wait.test.ts @@ -0,0 +1,121 @@ +/** + * #2308: supervisor DB-lock acquisition waits out a DEAD holder (e.g. the + * previous container's supervisor, whose row is cross-host and can only lapse + * via TTL) instead of one-shot exiting LOCK_HELD inside the TTL window. + * + * Pure-seam tests — no DB, no process.exit. The engine argument is never + * touched because every seam is injected. + */ + +import { describe, test, expect } from 'bun:test'; +import { hostname } from 'node:os'; +import { acquireSupervisorDbLockWithWait } from '../src/core/minions/supervisor.ts'; +import type { DbLockHandle, LockSnapshot } from '../src/core/db-lock.ts'; +import type { BrainEngine } from '../src/core/engine.ts'; + +const fakeEngine = {} as BrainEngine; + +const handle: DbLockHandle = { + id: 'gbrain-supervisor:default', + release: async () => {}, + refresh: async () => {}, +}; + +function snap(over: Partial = {}): LockSnapshot { + const now = Date.now(); + return { + id: 'gbrain-supervisor:default', + holder_pid: 424242, + holder_host: `dead-container-${hostname()}-not`, // never the local host + acquired_at: new Date(now - 120_000), + ttl_expires_at: new Date(now + 180_000), // TTL NOT expired — the bug window + age_ms: 120_000, + ttl_expired: false, + last_refreshed_at: new Date(now - 90_000), + ms_since_last_refresh: 90_000, + ...over, + }; +} + +describe('acquireSupervisorDbLockWithWait (#2308)', () => { + test('returns immediately when the first acquire succeeds', async () => { + let inspects = 0; + const got = await acquireSupervisorDbLockWithWait(fakeEngine, 'lock', 5, { + tryAcquire: async () => handle, + inspect: async () => { inspects++; return null; }, + sleep: async () => {}, + }); + expect(got).toBe(handle); + expect(inspects).toBe(0); + }); + + test('waits out a dead cross-host holder until the TTL-lapse takeover fires', async () => { + let attempts = 0; + let sleeps = 0; + const dead = snap(); // static heartbeat: never advances + const got = await acquireSupervisorDbLockWithWait(fakeEngine, 'lock', 5, { + tryAcquire: async () => (++attempts >= 3 ? handle : null), + inspect: async () => dead, + sleep: async () => { sleeps++; }, + retryMs: 1, + }); + expect(got).toBe(handle); + expect(attempts).toBe(3); + expect(sleeps).toBe(2); + }); + + test('fails fast when the holder is a live same-host PID', async () => { + let sleeps = 0; + const got = await acquireSupervisorDbLockWithWait(fakeEngine, 'lock', 5, { + tryAcquire: async () => null, + inspect: async () => snap({ holder_host: hostname() }), + // probe succeeds → PID alive + processKill: () => {}, + sleep: async () => { sleeps++; }, + }); + expect(got).toBeNull(); + expect(sleeps).toBe(0); + }); + + test('fails fast when the heartbeat ADVANCES between polls (live holder elsewhere)', async () => { + let inspects = 0; + const base = Date.now(); + const got = await acquireSupervisorDbLockWithWait(fakeEngine, 'lock', 5, { + tryAcquire: async () => null, + inspect: async () => { + inspects++; + // Heartbeat advances 60s per poll — an actively refreshing holder. + return snap({ last_refreshed_at: new Date(base + inspects * 60_000) }); + }, + sleep: async () => {}, + retryMs: 1, + }); + expect(got).toBeNull(); + expect(inspects).toBe(2); // second observation proves liveness + }); + + test('gives up after maxWaitMs when takeover never becomes possible', async () => { + let clock = 0; + const dead = snap(); + const got = await acquireSupervisorDbLockWithWait(fakeEngine, 'lock', 5, { + tryAcquire: async () => null, + inspect: async () => dead, + sleep: async () => { clock += 10_000; }, + now: () => clock, + retryMs: 10_000, + maxWaitMs: 35_000, + }); + expect(got).toBeNull(); + }); + + test('keeps polling when the row vanishes between inspects', async () => { + let attempts = 0; + const got = await acquireSupervisorDbLockWithWait(fakeEngine, 'lock', 5, { + tryAcquire: async () => (++attempts >= 2 ? handle : null), + inspect: async () => null, // row already gone + sleep: async () => {}, + retryMs: 1, + }); + expect(got).toBe(handle); + }); +}); diff --git a/test/write-through.test.ts b/test/write-through.test.ts index 238082a60..07b5c456a 100644 --- a/test/write-through.test.ts +++ b/test/write-through.test.ts @@ -189,3 +189,80 @@ describe('writePageThrough', () => { expect(files.some((f) => f.includes('.tmp.'))).toBe(false); }); }); + +// #2839: writePageThrough renders from the DB row, but fence writers +// (writeFactsToFence / forget) edit the on-disk `## Facts` fence FIRST and the +// DB body only catches up on the next sync. The write-through must merge those +// on-disk fence rows back instead of silently clobbering them. +describe('writePageThrough — facts fence preservation (#2839)', () => { + test('preserves on-disk fence rows the DB body has not caught up with', async () => { + await engine.setConfig('sync.repo_path', brainDir); + const slug = 'people/carol'; + await seedPage(slug); + + // First write-through creates the file from the row. + const first = await writePageThrough(engine, slug, { sourceId: 'default' }); + expect(first.written).toBe(true); + const filePath = first.path!; + + // Simulate a fence write landing on disk after the last import. + const { upsertFactRow } = await import('../src/core/facts-fence.ts'); + const { body: withFence } = upsertFactRow(fs.readFileSync(filePath, 'utf8'), { + claim: 'Deposited via fence write', + kind: 'fact', + confidence: 1.0, + visibility: 'world', + notability: 'high', + validFrom: '2026-01-01', + source: 'test', + }); + fs.writeFileSync(filePath, withFence, 'utf8'); + + // A put_page-style write-through re-renders from the (lagging) DB row... + const second = await writePageThrough(engine, slug, { sourceId: 'default' }); + expect(second.written).toBe(true); + + // ...but the fence row deposited on disk survives. + const after = fs.readFileSync(filePath, 'utf8'); + expect(after).toContain('Deposited via fence write'); + expect(after).toContain('## Facts'); + }); +}); + +describe('mergeDiskFactsFence (#2839)', () => { + const row = (rowNum: number, claim: string) => + `| ${rowNum} | ${claim} | fact | 1.0 | world | high | 2026-01-01 | | test | |`; + const fence = (rows: string[]) => [ + '', + '| # | claim | kind | confidence | visibility | notability | valid_from | valid_until | source | context |', + '|---|---|---|---|---|---|---|---|---|---|', + ...rows, + '', + ].join('\n'); + + test('unions rows by row_num, disk wins collisions', async () => { + const { mergeDiskFactsFence } = await import('../src/core/write-through.ts'); + const rendered = `# Page\n\n## Facts\n\n${fence([row(1, 'old claim one')])}\n`; + const disk = `# Page\n\n## Facts\n\n${fence([row(1, 'edited claim one'), row(2, 'new disk-only claim')])}\n`; + const merged = mergeDiskFactsFence(rendered, disk); + expect(merged).toContain('edited claim one'); + expect(merged).toContain('new disk-only claim'); + expect(merged).not.toContain('old claim one'); + }); + + test('appends a fence section when the rendered body has none', async () => { + const { mergeDiskFactsFence } = await import('../src/core/write-through.ts'); + const rendered = '# Page\n\nBody only.\n'; + const disk = `# Page\n\n## Facts\n\n${fence([row(1, 'disk fact')])}\n`; + const merged = mergeDiskFactsFence(rendered, disk); + expect(merged).toContain('Body only.'); + expect(merged).toContain('## Facts'); + expect(merged).toContain('disk fact'); + }); + + test('no-op when the disk body has no fence', async () => { + const { mergeDiskFactsFence } = await import('../src/core/write-through.ts'); + const rendered = '# Page\n\nRendered.\n'; + expect(mergeDiskFactsFence(rendered, '# Page\n\nNo fence here.\n')).toBe(rendered); + }); +}); diff --git a/test/zombie-reap.test.ts b/test/zombie-reap.test.ts index a9235aecf..3f26206e7 100644 --- a/test/zombie-reap.test.ts +++ b/test/zombie-reap.test.ts @@ -45,3 +45,88 @@ describe('installSigchldHandler', () => { expect(afterSecond).toBe(afterFirst); }); }); + +// #2443: PID-1 orphan reaper — /proc stat parsing + two-sweep confirmation. +import { + parseProcStatLine, + createOrphanSweeper, + installPid1OrphanReaper, +} from '../src/core/zombie-reap.ts'; + +describe('parseProcStatLine (#2443)', () => { + test('parses a plain zombie git child of PID 1', () => { + expect(parseProcStatLine('336 (git) Z 1 336 1 0 -1 4227084')).toEqual({ + pid: 336, + state: 'Z', + ppid: 1, + }); + }); + + test('handles comm with spaces and parens (splits on LAST close-paren)', () => { + expect(parseProcStatLine('42 (my (weird) prog) R 7 42 7')).toEqual({ + pid: 42, + state: 'R', + ppid: 7, + }); + }); + + test('returns null on garbage', () => { + expect(parseProcStatLine('')).toBeNull(); + expect(parseProcStatLine('not a stat line')).toBeNull(); + }); +}); + +describe('createOrphanSweeper (#2443)', () => { + test('reaps only pids seen in two consecutive sweeps', async () => { + const reaped: number[] = []; + let zombies: number[] = [10, 11]; + const sweeper = createOrphanSweeper(() => zombies, (pid) => reaped.push(pid)); + + await sweeper.sweep(); // first sighting — nothing reaped yet + expect(reaped).toEqual([]); + + zombies = [10, 12]; // 11 vanished (runtime reaped it), 12 is new + await sweeper.sweep(); + expect(reaped).toEqual([10]); // persisted across two sweeps → orphan + + zombies = [12]; + await sweeper.sweep(); + expect(reaped).toEqual([10, 12]); + }); + + test('a pid that disappears between sweeps is never reaped', async () => { + const reaped: number[] = []; + let zombies: number[] = [77]; + const sweeper = createOrphanSweeper(() => zombies, (pid) => reaped.push(pid)); + await sweeper.sweep(); + zombies = []; + await sweeper.sweep(); + expect(reaped).toEqual([]); + }); +}); + +describe('installPid1OrphanReaper (#2443)', () => { + test('no-op unless PID 1 on Linux', () => { + expect(installPid1OrphanReaper({ pid: process.pid, platform: 'linux' })).toBeNull(); + expect(installPid1OrphanReaper({ pid: 1, platform: 'darwin' })).toBeNull(); + }); + + test('installs an interval sweeper as PID 1 on Linux and reaps persistent zombies', async () => { + const reaped: number[] = []; + const stop = installPid1OrphanReaper({ + pid: 1, + platform: 'linux', + intervalMs: 5, + listZombieChildren: () => [99], + reapPid: (pid) => reaped.push(pid), + }); + expect(stop).not.toBeNull(); + try { + await new Promise((r) => setTimeout(r, 40)); + expect(reaped.length).toBeGreaterThan(0); + expect(reaped[0]).toBe(99); + } finally { + stop!(); + } + }); +});