v0.42.37.0 fix(jobs): reap stale locks, bound disconnect, complete cooperative-abort (#1972) (#2015)

* fix(jobs): reap stale dead-holder cycle/sync locks (#1972)

A crashed sync (OOM, recycle, SIGKILL) stranded its gbrain_cycle_locks row
until something contended for it — reclaim was on-contention only. Add a
host-scoped background reaper: reapDeadHolderLocks deletes locks whose holder
PID is provably dead on this host, scoped to the gbrain-sync:*/gbrain-cycle*
namespaces only (never elections/supervisor/reindex), with a snapshot-matched
delete (date_trunc on acquired_at) that is TOCTOU-safe against PID reuse.
Reuses isHolderDeadLocally (same-host + ESRCH + 60s grace). doctor --fix now
auto-reaps for no-autopilot brains. DRY: selectLockRows + shared mapper now
back inspectLock + listStaleLocks (killed the triplication).

* fix(db): bound pool disconnect so teardown can't eat CLI output (#1972)

pool.end() against PgBouncer transaction-mode never drained, so disconnect
blocked until the CLI's 10s force-exit fired and process.exit()'d mid-write,
truncating stdout (e.g. #1959's relational query returned empty). Add a
gbrain-owned endPoolBounded(pool): Promise.race of pool.end({timeout}) against
a hard timer, so teardown is bounded regardless of what postgres.js does and is
testable. connection-manager ends its direct + read pools concurrently so the
per-pool bounds don't stack. PGLite disconnect is unaffected.

* fix(cycle): complete cooperative-abort coverage + wire lock reaper (#1972)

v0.42.29 made only the embed phase honor the abort signal; a 24h pull still
showed force-evicts from a long non-embed phase ignoring it. Thread the signal
into every cycle-reachable long loop: extract (extractForSlugs + the full-walk
extractLinksFromDir/extractTimelineFromDir), extract_facts (per-page loop +
embed signal + the phantom-redirect 30s lock-retry), and consolidate's bucket
loop. Add a terminal abort check so an aborted cycle never stamps
last_full_cycle_at as a completed run (Codex #9). lint now yields + checks
abort every 200 pages (it's synchronous; the yield is what lets the signal
land). New phase-duration force-evict attribution log names any phase that
crosses the 30s deadline. Wire reapDeadHolderLocks at cycle start.

* chore: bump version and changelog (v0.42.38.0)

#1972 — stale-lock reaper, bounded pool disconnect, and complete
cooperative-abort coverage across cycle phases.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(key-files): current-state for the #1972 reaper, bounded disconnect, abort coverage

document-release: update db-lock.ts (reapDeadHolderLocks + selectLockRows DRY),
db.ts (endPoolBounded), and abort-check.ts (coverage now spans extract/
extract_facts/consolidate/lint + terminal guard) entries to current truth.

* test(isolation): fix shard-order flakes exposed by #1972's new test files

Adding 3 new test files reshuffled the hash-based shards, exposing two
pre-existing test-isolation bugs:

- cycle-consolidate.test.ts assumed the global legacy-embedding preload's
  1536-d gateway config still held at initSchema, but a co-sharded test that
  calls resetGateway() in teardown nulls it, so initSchema fell back to the
  1280-d default and built a halfvec(1280) facts column its 1536-d fixtures
  can't fill. Re-pin the legacy OpenAI/1536 config in beforeAll (the pattern
  legacy-embedding-preload.ts documents for 1536-d fixture tests).
- db-lock-heartbeat-takeover.test.ts (merged from master's #1794) mutated
  process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS raw, tripping check:test-isolation
  rule R1. Convert to withEnv().

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
2026-06-09 22:28:44 -07:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 1eb430a2df
commit 03ffc6ebdb
24 changed files with 809 additions and 138 deletions
+47
View File
@@ -152,3 +152,50 @@ describe('autopilot-cycle handler contract (v0.20.5)', () => {
expect(checkCalls).toBeGreaterThanOrEqual(6);
});
});
describe('#1972 — complete cooperative-abort coverage', () => {
test('aborted signal makes runCycle report partial + reason "aborted" (terminal guard)', async () => {
// phases:[] means no between-phase checkAborted fires, so execution reaches
// the TERMINAL guard (Codex #9). With a pre-aborted signal it must NOT
// report success — status 'partial', reason 'aborted'.
const { runCycle } = await import('../src/core/cycle.ts');
const abort = new AbortController();
abort.abort(new Error('timeout'));
const report = await runCycle(null, {
brainDir: '/nonexistent-for-test',
phases: [],
signal: abort.signal,
});
expect(report.status).toBe('partial');
expect(report.reason).toBe('aborted');
});
test('cycle.ts threads opts.signal into every long phase + guards the success stamp', async () => {
const fs = await import('fs');
const src = fs.readFileSync(new URL('../src/core/cycle.ts', import.meta.url), 'utf8');
const body = src.slice(src.indexOf('export async function runCycle'));
// Each long phase receives the signal.
expect(body).toContain('runPhaseExtract(engine, brainDir, dryRun, syncPagesAffected, opts.signal)');
expect(body).toMatch(/runPhaseExtractFacts\([^)]*opts\.signal\)/);
expect(body).toContain('signal: opts.signal'); // consolidate opts
expect(body).toContain('runPhaseLint(brainDir, dryRun, engine, opts.signal)');
// Reaper runs at cycle start.
expect(body).toContain('reapDeadHolderLocks(engine)');
// Terminal guard: the success stamp is gated on !aborted, and the report
// carries reason 'aborted'.
expect(body).toContain('!aborted');
expect(body).toContain("reason: 'aborted'");
// Phase-duration force-evict attribution log (T11).
expect(body).toContain('FORCE_EVICT_DEADLINE_MS');
});
test('every long phase core checks isAborted in its batch loop', async () => {
const fs = await import('fs');
const read = (p: string) => fs.readFileSync(new URL(p, import.meta.url), 'utf8');
expect(read('../src/commands/extract.ts')).toContain('if (isAborted(signal)) return;');
expect(read('../src/core/cycle/extract-facts.ts')).toContain('if (isAborted(opts.signal)) break;');
expect(read('../src/core/cycle/phases/consolidate.ts')).toContain('if (isAborted(opts.signal)) break;');
expect(read('../src/core/cycle/phantom-redirect.ts')).toContain('if (isAborted(signal)) break;');
expect(read('../src/commands/lint.ts')).toContain('if (isAborted(opts.signal)) break;');
});
});
+15
View File
@@ -10,6 +10,7 @@
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
import { configureGateway } from '../src/core/ai/gateway.ts';
import { runPhaseConsolidate } from '../src/core/cycle/phases/consolidate.ts';
let engine: PGLiteEngine;
@@ -17,6 +18,20 @@ let engine: PGLiteEngine;
beforeAll(async () => {
engine = new PGLiteEngine();
await engine.connect({});
// initSchema() bakes the facts.embedding dim from the gateway's configured
// embedding model; the default is now 1280-d (ZE). This file's fixtures are
// 1536-d, so pin the legacy 1536-d OpenAI config (matching
// test/helpers/legacy-embedding-preload.ts) right before initSchema. The
// global preload sets this, but a co-sharded test that calls resetGateway()
// in its teardown nulls it, leaving initSchema to fall back to the 1280-d
// default and build a halfvec(1280) column the 1536-d inserts can't fill.
// Re-pinning here makes the schema deterministic regardless of shard
// neighbors (surfaced when #1972's new test files reshuffled the shards).
configureGateway({
embedding_model: 'openai:text-embedding-3-large',
embedding_dimensions: 1536,
env: { ...process.env },
});
await engine.initSchema();
});
+7 -12
View File
@@ -16,6 +16,7 @@
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
import { hostname } from 'os';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
import { withEnv } from './helpers/with-env.ts';
import {
tryAcquireDbLock,
resolveStealGraceSeconds,
@@ -81,22 +82,16 @@ describe('resolveStealGraceSeconds', () => {
expect(resolveStealGraceSeconds(1)).toBe(60);
});
test('env override wins', () => {
process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = '123';
try {
test('env override wins', async () => {
await withEnv({ GBRAIN_LOCK_STEAL_GRACE_SECONDS: '123' }, async () => {
expect(resolveStealGraceSeconds(30)).toBe(123);
} finally {
delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS;
}
});
});
test('bad env override falls back to derived', () => {
process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = 'nope';
try {
test('bad env override falls back to derived', async () => {
await withEnv({ GBRAIN_LOCK_STEAL_GRACE_SECONDS: 'nope' }, async () => {
expect(resolveStealGraceSeconds(30)).toBe(600);
} finally {
delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS;
}
});
});
});
+162
View File
@@ -0,0 +1,162 @@
/**
* #1972 — host-scoped reaper for dead-holder sync/cycle locks.
*
* Covers:
* - reapDeadHolderLocks: namespace scope (sync/cycle only, NOT election/etc),
* same-host dead-PID reaped regardless of TTL, live/cross-host/within-grace
* kept. Uses the injectable process.kill seam so it's deterministic.
* - deleteLockRowExact: snapshot-matched delete (the TOCTOU defense) — a
* non-matching acquired_at is a no-op, the matching one deletes.
*/
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
import { hostname } from 'os';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
import {
reapDeadHolderLocks,
deleteLockRowExact,
inspectLock,
HOLDER_TAKEOVER_GRACE_MS,
type HolderLivenessOpts,
} from '../src/core/db-lock.ts';
let engine: PGLiteEngine;
beforeAll(async () => {
engine = new PGLiteEngine();
await engine.connect({ database_url: '' });
await engine.initSchema();
});
afterAll(async () => {
await engine.disconnect();
});
beforeEach(async () => {
await engine.executeRaw(`DELETE FROM gbrain_cycle_locks`);
});
const LOCAL = hostname();
const OLD_S = Math.ceil(HOLDER_TAKEOVER_GRACE_MS / 1000) + 60; // comfortably past grace
const YOUNG_S = 5;
/** Insert a lock row with a given age + namespace. ttlFuture controls whether
* the row is TTL-expired (to prove the reaper reaps dead holders regardless). */
async function seedLock(
id: string,
holderPid: number,
holderHost: string,
ageSeconds: number,
ttlFuture = true,
) {
const ttl = ttlFuture ? `NOW() + INTERVAL '10 minutes'` : `NOW() - INTERVAL '1 minute'`;
await engine.executeRaw(
`INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at)
VALUES ($1, $2, $3, NOW() - ($4 || ' seconds')::interval, ${ttl}, NOW() - ($4 || ' seconds')::interval)`,
[id, holderPid, holderHost, String(ageSeconds)],
);
}
/** process.kill seam: every pid is dead (ESRCH) except those in `live`. */
function killSeam(live: Set<number>): HolderLivenessOpts {
return {
localHost: LOCAL,
processKill: (pid: number) => {
if (live.has(pid)) return; // alive
const e = new Error('no such process') as NodeJS.ErrnoException;
e.code = 'ESRCH';
throw e;
},
};
}
async function lockIds(): Promise<string[]> {
const rows = await engine.executeRaw<{ id: string }>(`SELECT id FROM gbrain_cycle_locks ORDER BY id`);
return rows.map(r => r.id);
}
describe('reapDeadHolderLocks', () => {
test('reaps same-host dead-PID sync + cycle locks (regardless of TTL)', async () => {
await seedLock('gbrain-sync:src-a', 900001, LOCAL, OLD_S, /*ttlFuture*/ true); // dead, TTL NOT expired
await seedLock('gbrain-cycle', 900002, LOCAL, OLD_S, false); // dead, TTL expired
await seedLock('gbrain-cycle:src-b', 900003, LOCAL, OLD_S, true); // dead, TTL NOT expired
const { reaped, reapedIds } = await reapDeadHolderLocks(engine, killSeam(new Set()));
expect(reaped).toBe(3);
expect(reapedIds.sort()).toEqual(['gbrain-cycle', 'gbrain-cycle:src-b', 'gbrain-sync:src-a']);
expect(await lockIds()).toEqual([]);
});
test('keeps a live same-host holder', async () => {
await seedLock('gbrain-sync:src-live', process.pid, LOCAL, OLD_S);
const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set([process.pid])));
expect(reaped).toBe(0);
expect(await lockIds()).toEqual(['gbrain-sync:src-live']);
});
test('keeps a dead holder still within the PID-reuse grace window', async () => {
await seedLock('gbrain-sync:src-young', 900010, LOCAL, YOUNG_S);
const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set()));
expect(reaped).toBe(0);
expect(await lockIds()).toEqual(['gbrain-sync:src-young']);
});
test('keeps a cross-host holder (cannot probe a remote PID)', async () => {
await seedLock('gbrain-sync:src-xhost', 900011, 'a-different-host', OLD_S);
const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set()));
expect(reaped).toBe(0);
expect(await lockIds()).toEqual(['gbrain-sync:src-xhost']);
});
test('NEVER reaps a non-sync/cycle namespace, even with a dead PID (blast radius)', async () => {
await seedLock('gbrain-election:leader', 900020, LOCAL, OLD_S);
await seedLock('some-other-lock', 900021, LOCAL, OLD_S);
const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set()));
expect(reaped).toBe(0);
expect(await lockIds()).toEqual(['gbrain-election:leader', 'some-other-lock']);
});
test('empty table → {reaped:0}', async () => {
const r = await reapDeadHolderLocks(engine, killSeam(new Set()));
expect(r).toEqual({ reaped: 0, reapedIds: [] });
});
test('mixed set: reaps only the eligible rows', async () => {
await seedLock('gbrain-sync:dead', 900030, LOCAL, OLD_S); // reap
await seedLock('gbrain-sync:live', process.pid, LOCAL, OLD_S); // keep (live)
await seedLock('gbrain-cycle:xhost', 900031, 'other', OLD_S); // keep (cross-host)
await seedLock('gbrain-election:x', 900032, LOCAL, OLD_S); // keep (namespace)
const { reaped, reapedIds } = await reapDeadHolderLocks(engine, killSeam(new Set([process.pid])));
expect(reaped).toBe(1);
expect(reapedIds).toEqual(['gbrain-sync:dead']);
expect(await lockIds()).toEqual(['gbrain-cycle:xhost', 'gbrain-election:x', 'gbrain-sync:live']);
});
});
describe('deleteLockRowExact (snapshot-matched, TOCTOU defense)', () => {
test('no-op when acquired_at does not match; deletes when it does', async () => {
await seedLock('gbrain-sync:exact', 900040, LOCAL, OLD_S);
const snap = await inspectLock(engine, 'gbrain-sync:exact');
expect(snap).not.toBeNull();
// Simulate a reused-PID takeover: same id + pid, but a different (newer)
// acquired_at than the one the reaper snapshotted → must NOT delete.
const wrong = new Date(snap!.acquired_at.getTime() + 3_600_000);
const miss = await deleteLockRowExact(engine, 'gbrain-sync:exact', 900040, wrong);
expect(miss.deleted).toBe(false);
expect(await lockIds()).toEqual(['gbrain-sync:exact']);
// The matching snapshot deletes.
const hit = await deleteLockRowExact(engine, 'gbrain-sync:exact', 900040, snap!.acquired_at);
expect(hit.deleted).toBe(true);
expect(await lockIds()).toEqual([]);
});
test('no-op when holder_pid does not match', async () => {
await seedLock('gbrain-sync:pidguard', 900050, LOCAL, OLD_S);
const snap = await inspectLock(engine, 'gbrain-sync:pidguard');
const res = await deleteLockRowExact(engine, 'gbrain-sync:pidguard', 111111, snap!.acquired_at);
expect(res.deleted).toBe(false);
expect(await lockIds()).toEqual(['gbrain-sync:pidguard']);
});
});
+67
View File
@@ -0,0 +1,67 @@
/**
* #1972 (Codex #7) — cooperative abort on the cycle-reachable extract paths.
*
* The cycle calls runExtractCore with `slugs` defined (incremental →
* extractForSlugs) or undefined (full walk → extractLinksFromDir /
* extractTimelineFromDir). All three must bail promptly on a pre-aborted
* signal. Behavioral, dryRun-only (no DB writes, so no FK setup needed).
*/
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { mkdtempSync, writeFileSync, rmSync } from 'fs';
import { join } from 'path';
import { tmpdir } from 'os';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
import { runExtractCore } from '../src/commands/extract.ts';
let engine: PGLiteEngine;
let dir: string;
beforeAll(async () => {
engine = new PGLiteEngine();
await engine.connect({ database_url: '' });
await engine.initSchema();
dir = mkdtempSync(join(tmpdir(), 'extract-abort-'));
// a.md links to b; both exist so the wikilink resolves.
writeFileSync(join(dir, 'a.md'), '# A\n\nsee [[b]] for more.\n');
writeFileSync(join(dir, 'b.md'), '# B\n\nplain.\n');
});
afterAll(async () => {
await engine.disconnect();
rmSync(dir, { recursive: true, force: true });
});
function aborted(): AbortSignal {
const a = new AbortController();
a.abort(new Error('timeout'));
return a.signal;
}
describe('extract cooperative abort', () => {
test('incremental path (extractForSlugs) processes 0 pages when pre-aborted', async () => {
const r = await runExtractCore(engine, {
mode: 'links', dir, dryRun: true, slugs: ['a', 'b'], signal: aborted(),
});
expect(r.pages_processed).toBe(0);
});
test('incremental path processes all pages without a signal', async () => {
const r = await runExtractCore(engine, {
mode: 'links', dir, dryRun: true, slugs: ['a', 'b'],
});
expect(r.pages_processed).toBe(2);
});
test('full-walk path (extractLinksFromDir) creates 0 links when pre-aborted', async () => {
const r = await runExtractCore(engine, {
mode: 'links', dir, dryRun: true, signal: aborted(),
});
expect(r.links_created).toBe(0);
});
test('full-walk path resolves the wikilink without a signal', async () => {
const r = await runExtractCore(engine, { mode: 'links', dir, dryRun: true });
expect(r.links_created).toBeGreaterThanOrEqual(1);
});
});
@@ -31,7 +31,11 @@ describe('phantom-redirect lock contract', () => {
test('IRON-RULE: acquireLockWithRetry call passes syncLockId(sourceId)', () => {
// Banned: the bare `SYNC_LOCK_ID` constant slipped back in.
// Required: the per-source helper threaded with the active sourceId.
expect(SRC).toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*syncLockId\s*\(\s*sourceId\s*\)\s*\)/);
// #1972: the call now threads an optional abort signal as a third arg
// (acquireLockWithRetry(engine, syncLockId(sourceId), signal)). The IRON-RULE
// is unchanged: it must pass the per-source syncLockId(sourceId), never bare
// SYNC_LOCK_ID. Allow the optional trailing signal arg.
expect(SRC).toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*syncLockId\s*\(\s*sourceId\s*\)\s*(?:,\s*signal\s*)?\)/);
expect(SRC).not.toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*SYNC_LOCK_ID\s*\)/);
});
+40
View File
@@ -0,0 +1,40 @@
/**
* #1972 — gbrain-owned hard bound on pool teardown.
*
* The bug: `pool.end()` against PgBouncer transaction-mode never drains, so
* disconnect blocked until the CLI's 10s force-exit fired and truncated stdout.
* postgres.js's own `{ timeout }` is internal (a stub ignores it; it's not a
* guarantee we own), so `endPoolBounded` wraps every end in a Promise.race we
* control. These tests assert the bound is real (resolves even when `.end()`
* never settles) and that we still pass `{ timeout }` so a healthy drain is fast.
*/
import { describe, test, expect } from 'bun:test';
import { endPoolBounded, POOL_END_TIMEOUT_SECONDS } from '../src/core/db.ts';
describe('endPoolBounded', () => {
test('resolves fast when .end() settles quickly, forwarding { timeout }', async () => {
let calledWith: unknown;
const pool = { end: async (opts?: { timeout?: number }) => { calledWith = opts; } };
const t0 = Date.now();
await endPoolBounded(pool);
expect(Date.now() - t0).toBeLessThan(500);
expect(calledWith).toEqual({ timeout: POOL_END_TIMEOUT_SECONDS });
});
test('resolves within the gbrain bound even when .end() NEVER settles', async () => {
// This is the PgBouncer hang: .end() returns a promise that never resolves.
// The bare `await pool.end()` would hang until the CLI's 10s force-exit.
const pool = { end: () => new Promise<void>(() => { /* never resolves */ }) };
const t0 = Date.now();
await endPoolBounded(pool);
const elapsed = Date.now() - t0;
expect(elapsed).toBeGreaterThanOrEqual(POOL_END_TIMEOUT_SECONDS * 1000);
expect(elapsed).toBeLessThan(5000); // well under the CLI's 10s force-exit deadline
});
test('never throws when .end() rejects (teardown must not propagate)', async () => {
const pool = { end: async () => { throw new Error('pool boom'); } };
await expect(endPoolBounded(pool)).resolves.toBeUndefined();
});
});
@@ -81,7 +81,10 @@ describe('postgres-engine / module-singleton ownership (#1471)', () => {
const disconnect = stripComments(extractFn(DB_SRC, 'disconnect'));
const snapshotIdx = disconnect.search(/const\s+s\s*=\s*sql/);
const nullIdx = disconnect.search(/\bsql\s*=\s*null/);
const endIdx = disconnect.search(/\bawait\s+s\.end\s*\(/);
// #1972: the pool end is now wrapped in the gbrain-owned hard bound
// `endPoolBounded(s)` instead of a bare `s.end()`. The ordering contract is
// unchanged: snapshot + null the singleton BEFORE awaiting the end.
const endIdx = disconnect.search(/\bawait\s+(?:s\.end\s*\(|endPoolBounded\s*\(\s*s\b)/);
expect(snapshotIdx).toBeGreaterThanOrEqual(0);
expect(nullIdx).toBeGreaterThanOrEqual(0);
expect(endIdx).toBeGreaterThanOrEqual(0);