mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 22:15:33 +00:00
v0.42.5.0 fix(minions): RSS watchdog opacity + pooler-reap self-heal + silent lens backlog + cycle lint DB-disconnect (#1678) (#1735)
* fix(minions): self-identifying RSS watchdog + cgroup-aware default + pooler-reap self-heal (#1678) Problem 1: distinct WORKER_EXIT_RSS_WATCHDOG exit code + cause-keyed supervisor breaker (bypasses the stable-run reset that hid the 400x/24h loop) + rss_watchdog audit bucket + 80% soft-warn; cgroup-aware resolveDefaultMaxRssMb replaces the flat 2048 default at every spawn site. Problem 2: CONNECTION_ENDED classified retryable; postgres-engine sql getter throws a retryable error on a reaped instance pool instead of the misleading module-singleton fallthrough; promoteDelayed reconnect-retry; claim recovers on the next poll tick (no double-claim); lock-renewal tick reconnect-once dep. * feat(cycle): surface silent extract_atoms backlog + bounded --drain + fix lint clobbering the shared DB connection (#1678) Problem 3: extract_atoms_backlog doctor check + pack_gated skip marker + shared countExtractAtomsBacklog; `gbrain dream --phase extract_atoms --drain [--window N]` single-hold bounded drain (same cycleLockIdFor, rediscover each batch, reports remaining, exits non-zero while work remains). Also fixes a real production bug found via E2E: the cycle lint phase's resolveLintContentSanity created + disconnected a module-style engine that nulled the shared db singleton mid-cycle, breaking every later phase with "connect() has not been called". Lint now reuses the caller's live engine (cycle + Minion handlers thread it; standalone CLI keeps the create-own path). * chore: bump version and changelog (v0.41.39.0) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(#1678): pre-landing review — route transaction/withReservedConnection through the sql getter + drain treats failed count as incomplete Codex adversarial review findings: - #2: transaction(), withReservedConnection(), and one other site bypassed the v0.42.2.0 sql-getter self-heal via `this._sql || db.getConnection()`, so a reaped instance pool fell through to the module singleton there. Route all three through `this.sql` so they throw the retryable instance-pool error and recover consistently (MinionQueue.transaction hits this). - #4: `gbrain dream --drain` treated a null backlog count (query failure) as success via `remaining ?? 0`; now null exits EXIT_DRAIN_INCOMPLETE so automation never believes an unverified backlog drained. - #1 (claim orphan) + #3 (PGLite drain lock) documented as follow-ups in TODOS. * docs: document v0.42.2.0 #1678 modules + behavior in CLAUDE.md Adds Key Files entries for worker-exit-codes.ts, rss-default.ts, and extract-atoms-drain.ts, plus v0.42.2.0 annotations on worker.ts, child-worker-supervisor.ts, lock-renewal-tick.ts, and dream.ts. Regenerated llms-full.txt to match (test/build-llms.test.ts gate). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore: re-version v0.42.2.0 → v0.42.5.0 across VERSION/package.json/CHANGELOG/docs/comments Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
5911072aec
commit
766604dea0
@@ -180,14 +180,15 @@ describe('pruneOldBatchRetryAuditFiles — codex H-8 actual pruning', () => {
|
||||
});
|
||||
|
||||
test('no-op when audit dir does not exist (ENOENT)', async () => {
|
||||
// Point GBRAIN_AUDIT_DIR at a guaranteed-missing subdir of the per-test
|
||||
// tmpDir. Without this override the function reads the real ~/.gbrain/audit,
|
||||
// so the assertion flakes on any dev machine that already has a real
|
||||
// batch-retry-*.jsonl on disk (returns kept:1, not kept:0). Hermetic now,
|
||||
// matching this file's header contract.
|
||||
await withEnv({ GBRAIN_AUDIT_DIR: path.join(tmpDir, 'does-not-exist') }, async () => {
|
||||
// Isolate GBRAIN_AUDIT_DIR at a guaranteed-nonexistent path (CLAUDE.md R1).
|
||||
// Pre-fix this called pruneOld with no override, so on a real dev machine it
|
||||
// walked ~/.gbrain/audit — which may already hold this-week's batch-retry
|
||||
// files from other (un-isolated) suites, making `kept` non-zero and the test
|
||||
// flaky. Pointing at a missing subdir makes the ENOENT path deterministic.
|
||||
const missing = path.join(tmpDir, 'does-not-exist');
|
||||
await withEnv({ GBRAIN_AUDIT_DIR: missing }, async () => {
|
||||
const result = pruneOldBatchRetryAuditFiles(30, new Date());
|
||||
// The function never throws on a missing dir; it returns the empty result.
|
||||
// The function never throws on a missing dir; returns the empty result.
|
||||
expect(result).toEqual({ removed: 0, kept: 0 });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
* static-shape regressions read the source file and pin the load-bearing
|
||||
* constants:
|
||||
*
|
||||
* - `--max-rss 2048` is passed to the worker (incident-driving default)
|
||||
* - the worker is spawned with an auto-sized `--max-rss` (issue #1678)
|
||||
* - `maxCrashes: 5` matches the prior `crashCount >= 5` give-up rule
|
||||
* - The autopilot composes ChildWorkerSupervisor (not the legacy
|
||||
* inline `child.on('exit')` loop)
|
||||
@@ -50,12 +50,15 @@ describe('autopilot.ts ↔ ChildWorkerSupervisor wiring', () => {
|
||||
expect(AUTOPILOT_SRC).not.toContain('STABLE_RUN_RESET_MS');
|
||||
});
|
||||
|
||||
it("constructs ChildWorkerSupervisor with --max-rss 2048", () => {
|
||||
// The worker spawn args must include both flag tokens in argv order.
|
||||
// This is the incident-driving default; changing it without a deliberate
|
||||
// decision would regress the workaround for VmRSS inflation.
|
||||
expect(AUTOPILOT_SRC).toContain("'--max-rss', '2048'");
|
||||
it("spawns the worker with an auto-sized --max-rss (issue #1678)", () => {
|
||||
// Post-v0.41.39.0 the flat 2048 default is gone: autopilot resolves
|
||||
// resolveDefaultMaxRssMb() (cgroup-aware) and passes it as the cap. The
|
||||
// argv must still carry the --max-rss flag token + the resolved value.
|
||||
expect(AUTOPILOT_SRC).toContain("resolveDefaultMaxRssMb");
|
||||
expect(AUTOPILOT_SRC).toContain("'--max-rss', String(autopilotMaxRssMb)");
|
||||
expect(AUTOPILOT_SRC).toContain("'jobs', 'work'");
|
||||
// The footgun literal must NOT come back.
|
||||
expect(AUTOPILOT_SRC).not.toContain("'--max-rss', '2048'");
|
||||
});
|
||||
|
||||
it("constructs ChildWorkerSupervisor with maxCrashes: 5", () => {
|
||||
|
||||
@@ -55,6 +55,9 @@ async function runUntilTerminal(
|
||||
cleanRestartWindowMs: number;
|
||||
cleanRestartBudgetBackoffMs: number;
|
||||
stableRunResetMs: number;
|
||||
watchdogLoopBudget: number;
|
||||
watchdogLoopWindowMs: number;
|
||||
watchdogBackoffMs: number;
|
||||
_now: () => number;
|
||||
stopAfterEvents: number; // safety net so a buggy test can't hang
|
||||
}>,
|
||||
@@ -73,6 +76,9 @@ async function runUntilTerminal(
|
||||
cleanRestartWindowMs: overrides.cleanRestartWindowMs,
|
||||
cleanRestartBudgetBackoffMs: overrides.cleanRestartBudgetBackoffMs,
|
||||
stableRunResetMs: overrides.stableRunResetMs,
|
||||
watchdogLoopBudget: overrides.watchdogLoopBudget,
|
||||
watchdogLoopWindowMs: overrides.watchdogLoopWindowMs,
|
||||
watchdogBackoffMs: overrides.watchdogBackoffMs,
|
||||
_now: overrides._now,
|
||||
isStopping: () => stopping,
|
||||
onMaxCrashesExceeded: (count, max) => {
|
||||
@@ -407,4 +413,61 @@ esac
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// issue #1678: RSS-watchdog exits (code 12) are cause-keyed and must NOT
|
||||
// route through the generic crash path — the >5-min stable-run reset would
|
||||
// defeat max_crashes and the 400×/24h loop would never stop being silent.
|
||||
describe('rss_watchdog breaker (issue #1678)', () => {
|
||||
it('code=12 is labeled rss_watchdog and never increments crashCount', async () => {
|
||||
const h = makeHarness('wd-nocrash', 'exit 12');
|
||||
try {
|
||||
const { events, maxCrashesFired } = await runUntilTerminal(h, {
|
||||
maxCrashes: 3,
|
||||
_backoffFloorMs: 1,
|
||||
stopAfterEvents: 18, // ~6 spawn/exit/backoff triples
|
||||
});
|
||||
const exited = events.filter(
|
||||
(e): e is Extract<ChildSupervisorEvent, { kind: 'worker_exited' }> =>
|
||||
e.kind === 'worker_exited',
|
||||
);
|
||||
// Looped well past maxCrashes WITHOUT tripping it — the whole point.
|
||||
expect(maxCrashesFired).toBeNull();
|
||||
expect(exited.length).toBeGreaterThan(3);
|
||||
for (const e of exited) {
|
||||
expect(e.code).toBe(12);
|
||||
expect(e.likelyCause).toBe('rss_watchdog');
|
||||
expect(e.crashCount).toBe(0); // never counted as a crash
|
||||
}
|
||||
} finally {
|
||||
h.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it('emits rss_watchdog_loop health_warn once the window budget is exceeded', async () => {
|
||||
const h = makeHarness('wd-loop', 'exit 12');
|
||||
try {
|
||||
const { events } = await runUntilTerminal(h, {
|
||||
maxCrashes: 99,
|
||||
_backoffFloorMs: 1,
|
||||
watchdogLoopBudget: 2,
|
||||
watchdogLoopWindowMs: 600_000,
|
||||
stopAfterEvents: 24,
|
||||
});
|
||||
const warns = events.filter(
|
||||
(e): e is Extract<ChildSupervisorEvent, { kind: 'health_warn' }> =>
|
||||
e.kind === 'health_warn' && e.reason === 'rss_watchdog_loop',
|
||||
);
|
||||
// Budget=2 → the 3rd+ watchdog exit in-window fires the loud alert.
|
||||
expect(warns.length).toBeGreaterThan(0);
|
||||
expect(warns[0].count).toBeGreaterThan(2);
|
||||
// And every backoff after a watchdog exit is reason=rss_watchdog.
|
||||
const wdBackoffs = events.filter(
|
||||
(e) => e.kind === 'backoff' && e.reason === 'rss_watchdog',
|
||||
);
|
||||
expect(wdBackoffs.length).toBeGreaterThan(0);
|
||||
} finally {
|
||||
h.cleanup();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -348,6 +348,7 @@ describe('BATCH_AUDIT_SITES typed enum + isBatchAuditSite guard (D10c codex)', (
|
||||
'sync.import_file',
|
||||
'reindex.markdown', 'reindex.multimodal',
|
||||
'backfill.outer',
|
||||
'minion-lock',
|
||||
]);
|
||||
expect(new Set<string>([...BATCH_AUDIT_SITES])).toEqual(expected);
|
||||
});
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
/**
|
||||
* issue #1678 — extract_atoms backlog count + doctor check.
|
||||
*
|
||||
* Pins:
|
||||
* - countExtractAtomsBacklog counts eligible-but-unextracted pages (scoped +
|
||||
* brain-wide) and excludes pages that already have an atom (NOT EXISTS).
|
||||
* - computeExtractAtomsBacklogCheck WARNs with a `--drain` hint when the pack
|
||||
* doesn't run the phase and the backlog is real; OK at 0.
|
||||
*
|
||||
* Real in-memory PGLite (canonical block, R3+R4). GBRAIN_HOME is pointed at an
|
||||
* empty tmpdir for the doctor-check cases so packDeclaresPhase resolves the
|
||||
* bundled base pack (which does NOT declare extract_atoms) deterministically,
|
||||
* independent of the developer's real ~/.gbrain config.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { mkdtempSync } from 'fs';
|
||||
import { tmpdir } from 'os';
|
||||
import { join } from 'path';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
import { withEnv } from './helpers/with-env.ts';
|
||||
import { countExtractAtomsBacklog } from '../src/core/cycle/extract-atoms.ts';
|
||||
import { computeExtractAtomsBacklogCheck } from '../src/commands/doctor.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
const EMPTY_HOME = mkdtempSync(join(tmpdir(), 'gbrain-xa-backlog-home-'));
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
const BODY = 'x'.repeat(600); // >= MIN_PAGE_CHARS_FOR_EXTRACTION (500)
|
||||
|
||||
async function seedArticle(slug: string) {
|
||||
return engine.putPage(slug, { type: 'article', title: slug, compiled_truth: BODY });
|
||||
}
|
||||
|
||||
describe('countExtractAtomsBacklog (issue #1678)', () => {
|
||||
it('counts eligible pages with no atom (scoped + brain-wide)', async () => {
|
||||
await seedArticle('article-a');
|
||||
await seedArticle('article-b');
|
||||
await seedArticle('article-c');
|
||||
expect(await countExtractAtomsBacklog(engine)).toBe(3);
|
||||
expect(await countExtractAtomsBacklog(engine, 'default')).toBe(3);
|
||||
});
|
||||
|
||||
it('excludes a page that already has a matching atom (NOT EXISTS)', async () => {
|
||||
const p = await seedArticle('article-x');
|
||||
const h16 = (p.content_hash ?? '').slice(0, 16);
|
||||
expect(h16.length).toBe(16);
|
||||
await engine.putPage('atoms/a1', {
|
||||
type: 'atom',
|
||||
title: 'a1',
|
||||
compiled_truth: 'an extracted nugget',
|
||||
frontmatter: { source_hash: h16 },
|
||||
});
|
||||
expect(await countExtractAtomsBacklog(engine)).toBe(0);
|
||||
});
|
||||
|
||||
it('ignores short pages and dream-generated pages', async () => {
|
||||
await engine.putPage('article-short', { type: 'article', title: 's', compiled_truth: 'too short' });
|
||||
await engine.putPage('article-dream', {
|
||||
type: 'article', title: 'd', compiled_truth: BODY,
|
||||
frontmatter: { dream_generated: 'true' },
|
||||
});
|
||||
expect(await countExtractAtomsBacklog(engine)).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('computeExtractAtomsBacklogCheck (issue #1678)', () => {
|
||||
it('OK with no backlog', async () => {
|
||||
const check = await withEnv({ GBRAIN_HOME: EMPTY_HOME }, () =>
|
||||
computeExtractAtomsBacklogCheck(engine));
|
||||
expect(check.status).toBe('ok');
|
||||
expect((check.details as { backlog: number }).backlog).toBe(0);
|
||||
});
|
||||
|
||||
it('WARNs with a --drain hint when the pack does not run the phase and backlog > 10', async () => {
|
||||
for (let i = 0; i < 11; i++) await seedArticle(`article-${i}`);
|
||||
const check = await withEnv({ GBRAIN_HOME: EMPTY_HOME }, () =>
|
||||
computeExtractAtomsBacklogCheck(engine));
|
||||
expect(check.status).toBe('warn');
|
||||
expect(check.message).toContain('--drain');
|
||||
expect((check.details as { pack_declares_phase: boolean }).pack_declares_phase).toBe(false);
|
||||
expect((check.details as { known_approximation: string }).known_approximation).toContain('page backlog only');
|
||||
});
|
||||
});
|
||||
@@ -105,4 +105,28 @@ describe('dream CLI flag wiring', () => {
|
||||
expect(dreamSrc).toContain('gbrain sources restore');
|
||||
});
|
||||
});
|
||||
|
||||
// issue #1678 — --drain bounded backlog drain wiring (structural).
|
||||
describe('--drain wiring', () => {
|
||||
test('declares --drain and --window flags', () => {
|
||||
expect(dreamSrc).toContain("'--drain'");
|
||||
expect(dreamSrc).toContain("'--window'");
|
||||
expect(dreamSrc).toContain('windowSeconds');
|
||||
});
|
||||
|
||||
test('--drain defaults to extract_atoms and rejects other phases', () => {
|
||||
expect(dreamSrc).toContain("phase = 'extract_atoms'");
|
||||
expect(dreamSrc).toContain('--drain currently supports only --phase extract_atoms');
|
||||
});
|
||||
|
||||
test('drain holds the same cycle lock id (contends with the routine cycle)', () => {
|
||||
expect(dreamSrc).toContain('cycleLockIdFor(resolvedSourceId)');
|
||||
expect(dreamSrc).toContain('withRefreshingLock');
|
||||
});
|
||||
|
||||
test('drain reports remaining + exits non-zero when incomplete', () => {
|
||||
expect(dreamSrc).toContain('EXIT_DRAIN_INCOMPLETE');
|
||||
expect(dreamSrc).toContain('cycle_already_running');
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
+8
-12
@@ -29,7 +29,7 @@ mock.module('../../src/core/embedding.ts', () => ({
|
||||
currentEmbeddingSignature: () => 'test:model:1536',
|
||||
}));
|
||||
|
||||
const { runCycle } = await import('../../src/core/cycle.ts');
|
||||
const { runCycle, ALL_PHASES } = await import('../../src/core/cycle.ts');
|
||||
|
||||
const skip = !hasDatabase();
|
||||
const describeE2E = skip ? describe.skip : describe;
|
||||
@@ -99,17 +99,13 @@ describeE2E('E2E: runCycle against real Postgres', () => {
|
||||
});
|
||||
|
||||
expect(report.schema_version).toBe('1');
|
||||
// Cycle ran all 16 phases (or skipped the ones that don't support dry-run).
|
||||
// Phase history:
|
||||
// v0.23 = 8 phases (lint → backlinks → sync → synthesize → extract → patterns → embed → orphans)
|
||||
// v0.26.5 = 9 (added `purge` after orphans)
|
||||
// v0.29 = 10 (added `recompute_emotional_weight` between patterns and embed)
|
||||
// v0.31 = 11 (added `consolidate` between recompute_emotional_weight and embed)
|
||||
// v0.32.2 = 12 (added `extract_facts` between extract and patterns)
|
||||
// v0.33.3 = 13 (added `resolve_symbol_edges` between extract_facts and patterns)
|
||||
// v0.36.1.0 = 16 (added propose_takes + grade_takes + calibration_profile — hindsight calibration wave)
|
||||
// v0.39.0.0 = 17 (added `schema-suggest` between orphans and purge — T12 schema cathedral)
|
||||
expect(report.phases.length).toBe(19); // v0.41: +extract_atoms, +synthesize_concepts
|
||||
// Every phase in ALL_PHASES pushes exactly one result (pack-gated phases
|
||||
// like extract_atoms / synthesize_concepts push a 'skipped' result), so
|
||||
// the full dry-run cycle's phase count always equals ALL_PHASES.length.
|
||||
// Assert against the live constant rather than a hardcoded number so this
|
||||
// doesn't go stale every time a phase is added (it drifted to 19 while the
|
||||
// real count was 20 — the conversation_facts_backfill phase was missed).
|
||||
expect(report.phases.length).toBe(ALL_PHASES.length);
|
||||
|
||||
// Nothing got written.
|
||||
const afterPages = await conn.unsafe(`SELECT count(*)::int AS n FROM pages`);
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
/**
|
||||
* issue #1678 — bounded single-hold extract_atoms drain loop.
|
||||
*
|
||||
* Pure-over-injected-deps, so no DB / LLM / lock primitive. Pins:
|
||||
* - drains to empty (rediscovers each batch via countRemaining), stops 'drained'
|
||||
* - the wallclock window bounds the loop, stops 'window' with remaining > 0
|
||||
* - a zero-progress batch stops the loop (no hot loop burning budget)
|
||||
* - a busy lock (withLock throws) propagates so the caller reports skipped
|
||||
*/
|
||||
|
||||
import { describe, it, expect } from 'bun:test';
|
||||
import {
|
||||
runExtractAtomsDrain,
|
||||
type ExtractAtomsDrainDeps,
|
||||
} from '../src/core/cycle/extract-atoms-drain.ts';
|
||||
|
||||
function seq(values: Array<number | null>): () => Promise<number | null> {
|
||||
let i = 0;
|
||||
return async () => values[Math.min(i++, values.length - 1)];
|
||||
}
|
||||
|
||||
const passThroughLock: ExtractAtomsDrainDeps['withLock'] = (work) => work();
|
||||
|
||||
describe('runExtractAtomsDrain (issue #1678)', () => {
|
||||
it('drains to empty and reports stopped=drained', async () => {
|
||||
let batches = 0;
|
||||
const result = await runExtractAtomsDrain(
|
||||
{
|
||||
withLock: passThroughLock,
|
||||
countRemaining: seq([3, 2, 1, 0, 0]),
|
||||
runBatch: async () => { batches++; return { extracted: 1, skipped: 0 }; },
|
||||
now: () => 0,
|
||||
},
|
||||
{ windowMs: 1_000_000 },
|
||||
);
|
||||
expect(result.stopped).toBe('drained');
|
||||
expect(result.remaining).toBe(0);
|
||||
expect(result.batches).toBe(3);
|
||||
expect(result.extracted).toBe(3);
|
||||
expect(batches).toBe(3);
|
||||
});
|
||||
|
||||
it('stops at the wallclock window with remaining > 0', async () => {
|
||||
// SYNC stepping clock: now() #1 sets deadline (0+100=100); the while-check
|
||||
// then sees 50, 50 (two batches), then 999999 → past deadline → stop.
|
||||
const times = [0, 50, 50, 999_999];
|
||||
let ti = 0;
|
||||
const now = () => times[Math.min(ti++, times.length - 1)];
|
||||
const result = await runExtractAtomsDrain(
|
||||
{
|
||||
withLock: passThroughLock,
|
||||
countRemaining: async () => 5, // never drains
|
||||
runBatch: async () => ({ extracted: 1, skipped: 0 }),
|
||||
now,
|
||||
},
|
||||
{ windowMs: 100 },
|
||||
);
|
||||
expect(result.stopped).toBe('window');
|
||||
expect(result.remaining).toBe(5);
|
||||
expect(result.batches).toBe(2);
|
||||
});
|
||||
|
||||
it('stops on a zero-progress batch (no hot loop)', async () => {
|
||||
let batches = 0;
|
||||
const result = await runExtractAtomsDrain(
|
||||
{
|
||||
withLock: passThroughLock,
|
||||
countRemaining: async () => 5,
|
||||
runBatch: async () => { batches++; return { extracted: 0, skipped: 0 }; },
|
||||
now: () => 0,
|
||||
},
|
||||
{ windowMs: 1_000_000 },
|
||||
);
|
||||
expect(result.stopped).toBe('no_progress');
|
||||
expect(batches).toBe(1);
|
||||
expect(result.remaining).toBe(5);
|
||||
});
|
||||
|
||||
it('propagates a busy-lock error (caller reports cycle_already_running)', async () => {
|
||||
class FakeBusy extends Error {}
|
||||
await expect(
|
||||
runExtractAtomsDrain(
|
||||
{
|
||||
withLock: () => { throw new FakeBusy('held'); },
|
||||
countRemaining: async () => 5,
|
||||
runBatch: async () => ({ extracted: 1, skipped: 0 }),
|
||||
now: () => 0,
|
||||
},
|
||||
{ windowMs: 1000 },
|
||||
),
|
||||
).rejects.toThrow('held');
|
||||
});
|
||||
|
||||
it('respects maxBatches as a belt-and-suspenders cap', async () => {
|
||||
let batches = 0;
|
||||
const result = await runExtractAtomsDrain(
|
||||
{
|
||||
withLock: passThroughLock,
|
||||
countRemaining: async () => 999, // never drains
|
||||
runBatch: async () => { batches++; return { extracted: 1, skipped: 0 }; },
|
||||
now: () => 0, // window never elapses
|
||||
},
|
||||
{ windowMs: 1_000_000, maxBatches: 4 },
|
||||
);
|
||||
expect(result.stopped).toBe('max_batches');
|
||||
expect(batches).toBe(4);
|
||||
});
|
||||
});
|
||||
Vendored
+6
@@ -40,6 +40,11 @@ const backoffFloor = parseInt(process.env.SUP_BACKOFF_FLOOR_MS ?? '1', 10);
|
||||
const healthInterval = parseInt(process.env.SUP_HEALTH_INTERVAL_MS ?? '999999', 10);
|
||||
const allowShellJobs = process.env.SUP_ALLOW_SHELL_JOBS === '1';
|
||||
const queueName = process.env.SUP_QUEUE ?? 'default';
|
||||
// SUP_MAX_RSS: when set, pin an explicit watchdog cap (tests the passthrough
|
||||
// path). When unset, MinionSupervisor auto-sizes cgroup-aware (issue #1678).
|
||||
const maxRssExplicit = process.env.SUP_MAX_RSS !== undefined
|
||||
? parseInt(process.env.SUP_MAX_RSS, 10)
|
||||
: undefined;
|
||||
|
||||
if (process.env.SUP_AUDIT_DIR) {
|
||||
process.env.GBRAIN_AUDIT_DIR = process.env.SUP_AUDIT_DIR;
|
||||
@@ -57,6 +62,7 @@ const supervisor = new MinionSupervisor(mockEngine as BrainEngine, {
|
||||
allowShellJobs,
|
||||
json: true,
|
||||
_backoffFloorMs: backoffFloor,
|
||||
...(maxRssExplicit !== undefined ? { maxRssMb: maxRssExplicit } : {}),
|
||||
onEvent: (emission) => writeSupervisorEvent(emission, supervisorPid),
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
/**
|
||||
* issue #1678 — lint must REUSE a caller-provided engine for the
|
||||
* content-sanity DB-plane config lift, never create + disconnect its own.
|
||||
*
|
||||
* The bug: resolveLintContentSanity created a module-style engine
|
||||
* (createEngine without poolSize wraps the db.ts singleton) and disconnect()ed
|
||||
* it, which cascaded to db.disconnect() and NULLED the shared singleton the
|
||||
* cycle's lint phase depends on — breaking every subsequent cycle phase with a
|
||||
* misleading "connect() has not been called". When the caller passes a live
|
||||
* engine, lint must use it directly with zero connection churn.
|
||||
*
|
||||
* Hermetic: a fake BrainEngine that records disconnect() calls + serves
|
||||
* getConfig. No real DB.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'bun:test';
|
||||
import { mkdtempSync, writeFileSync, rmSync } from 'fs';
|
||||
import { tmpdir } from 'os';
|
||||
import { join } from 'path';
|
||||
import { runLintCore } from '../src/commands/lint.ts';
|
||||
import type { BrainEngine } from '../src/core/engine.ts';
|
||||
|
||||
let dir: string;
|
||||
beforeEach(() => {
|
||||
dir = mkdtempSync(join(tmpdir(), 'lint-shared-engine-'));
|
||||
writeFileSync(join(dir, 'a.md'), '---\ntype: note\ntitle: A\n---\n\nSome content.\n');
|
||||
});
|
||||
afterEach(() => {
|
||||
try { rmSync(dir, { recursive: true, force: true }); } catch { /* noop */ }
|
||||
});
|
||||
|
||||
describe('runLintCore engine reuse (issue #1678)', () => {
|
||||
it('reuses a provided engine for the content-sanity lift and NEVER disconnects it', async () => {
|
||||
const state = { disconnects: 0, connects: 0, getConfigCalls: 0 };
|
||||
const engine = {
|
||||
kind: 'postgres' as const,
|
||||
getConfig: async () => { state.getConfigCalls++; return null; },
|
||||
connect: async () => { state.connects++; },
|
||||
disconnect: async () => { state.disconnects++; },
|
||||
} as unknown as BrainEngine;
|
||||
|
||||
await runLintCore({ target: dir, fix: false, dryRun: true, engine });
|
||||
|
||||
// The load-bearing assertion: the shared engine was used (getConfig hit)
|
||||
// but NEVER disconnected and NEVER re-connected — no connection churn that
|
||||
// could null a shared singleton mid-cycle.
|
||||
expect(state.getConfigCalls).toBeGreaterThan(0);
|
||||
expect(state.disconnects).toBe(0);
|
||||
expect(state.connects).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,43 @@
|
||||
/**
|
||||
* issue #1678 — the `PostgresEngine.sql` getter must not fall through to the
|
||||
* never-connected module singleton when an INSTANCE pool's _sql went null
|
||||
* (mid-process disconnect, or a reaped pooler socket). Pre-fix it threw the
|
||||
* misleading "connect() has not been called"; post-fix it throws a tailored
|
||||
* RETRYABLE error so withRetry+reconnect rebuilds the pool and recovers.
|
||||
*
|
||||
* Pure: pokes the private fields and reads the synchronous getter; no real DB.
|
||||
*/
|
||||
|
||||
import { describe, it, expect } from 'bun:test';
|
||||
import { PostgresEngine } from '../src/core/postgres-engine.ts';
|
||||
import { isRetryableConnError } from '../src/core/retry-matcher.ts';
|
||||
|
||||
describe('PostgresEngine.sql getter self-heal (issue #1678)', () => {
|
||||
it('instance-pool + null _sql throws a RETRYABLE error naming the reaped pool', () => {
|
||||
const e = new PostgresEngine();
|
||||
(e as unknown as { _connectionStyle: string })._connectionStyle = 'instance';
|
||||
(e as unknown as { _sql: unknown })._sql = null;
|
||||
|
||||
let thrown: unknown;
|
||||
try {
|
||||
// accessing the getter triggers the throw
|
||||
void e.sql;
|
||||
} catch (err) {
|
||||
thrown = err;
|
||||
}
|
||||
expect(thrown).toBeDefined();
|
||||
// Must be classified retryable so the lock/batch retry paths reconnect.
|
||||
expect(isRetryableConnError(thrown)).toBe(true);
|
||||
// Must NOT be the misleading legacy message.
|
||||
const msg = (thrown as Error).message;
|
||||
expect(msg).toContain('instance connection pool');
|
||||
expect(msg).not.toContain('connect() has not been called');
|
||||
});
|
||||
|
||||
it('a live instance _sql is returned directly (no throw)', () => {
|
||||
const e = new PostgresEngine();
|
||||
const fakeSql = { tag: 'live-pool' };
|
||||
(e as unknown as { _sql: unknown })._sql = fakeSql;
|
||||
expect(e.sql as unknown).toBe(fakeSql);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,70 @@
|
||||
/**
|
||||
* issue #1678 — Minion hot-path lock recovery contract.
|
||||
*
|
||||
* - promoteDelayed (idempotent) self-heals: a reaped-socket CONNECTION_ENDED
|
||||
* triggers a reconnect + retry against a fresh pool.
|
||||
* - claim does NOT retry inline (Codex #1): blind-retrying a claim whose
|
||||
* UPDATE...RETURNING may have committed could double-claim a job. The error
|
||||
* propagates; the worker poll loop reconnects + re-claims on the next tick.
|
||||
*
|
||||
* Hermetic: a fake BrainEngine whose executeRaw is scripted; no real DB.
|
||||
*/
|
||||
|
||||
import { describe, it, expect } from 'bun:test';
|
||||
import { tmpdir } from 'os';
|
||||
import { join } from 'path';
|
||||
import { MinionQueue } from '../src/core/minions/queue.ts';
|
||||
import { withEnv } from './helpers/with-env.ts';
|
||||
|
||||
function connEndedError(): Error & { code: string } {
|
||||
const e = new Error('write CONNECTION_ENDED localhost:6543') as Error & { code: string };
|
||||
e.code = 'CONNECTION_ENDED';
|
||||
return e;
|
||||
}
|
||||
|
||||
const AUDIT_DIR = join(tmpdir(), `gbrain-queue-lock-retry-${process.pid}-${Date.now()}`);
|
||||
// Fast retry + isolated audit dir so the test doesn't sleep ~1s or pollute ~/.gbrain.
|
||||
const FAST_ENV = {
|
||||
GBRAIN_BULK_RETRY_BASE_MS: '1',
|
||||
GBRAIN_BULK_RETRY_MAX_MS: '2',
|
||||
GBRAIN_AUDIT_DIR: AUDIT_DIR,
|
||||
};
|
||||
|
||||
describe('MinionQueue lock-path recovery (issue #1678)', () => {
|
||||
it('promoteDelayed reconnects + retries on a reaped-socket error', async () => {
|
||||
await withEnv(FAST_ENV, async () => {
|
||||
let calls = 0;
|
||||
let reconnects = 0;
|
||||
const engine = {
|
||||
kind: 'postgres',
|
||||
executeRaw: async () => {
|
||||
calls++;
|
||||
if (calls === 1) throw connEndedError();
|
||||
return [];
|
||||
},
|
||||
reconnect: async () => { reconnects++; },
|
||||
} as unknown as ConstructorParameters<typeof MinionQueue>[0];
|
||||
|
||||
const q = new MinionQueue(engine);
|
||||
const out = await q.promoteDelayed();
|
||||
expect(out).toEqual([]);
|
||||
expect(calls).toBe(2); // first attempt threw, retry succeeded
|
||||
expect(reconnects).toBe(1); // reconnect fired between attempts
|
||||
});
|
||||
});
|
||||
|
||||
it('claim does NOT retry inline on a reaped-socket error (Codex #1 double-claim guard)', async () => {
|
||||
await withEnv(FAST_ENV, async () => {
|
||||
let calls = 0;
|
||||
const engine = {
|
||||
kind: 'postgres',
|
||||
executeRaw: async () => { calls++; throw connEndedError(); },
|
||||
reconnect: async () => {},
|
||||
} as unknown as ConstructorParameters<typeof MinionQueue>[0];
|
||||
|
||||
const q = new MinionQueue(engine);
|
||||
await expect(q.claim('tok', 1000, 'default', ['sync'])).rejects.toThrow('CONNECTION_ENDED');
|
||||
expect(calls).toBe(1); // exactly one attempt — no inline retry
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -75,6 +75,24 @@ describe('isRetryableConnError', () => {
|
||||
test('does not match arbitrary errors', () => {
|
||||
expect(isRetryableConnError(new Error('something else'))).toBe(false);
|
||||
});
|
||||
|
||||
// issue #1678: postgres.js's transaction-mode pooler reaps idle sockets and
|
||||
// throws errors carrying `code: 'CONNECTION_ENDED'` (a library code, not an
|
||||
// 08xxx SQLSTATE). Must be retryable via BOTH the code and the message form.
|
||||
test('matches CONNECTION_ENDED via code', () => {
|
||||
expect(isRetryableConnError(pgError('CONNECTION_ENDED', 'write CONNECTION_ENDED'))).toBe(true);
|
||||
});
|
||||
|
||||
test('matches CONNECTION_ENDED via message even without the code', () => {
|
||||
expect(isRetryableConnError(new Error('write CONNECTION_ENDED localhost:6543'))).toBe(true);
|
||||
});
|
||||
|
||||
// The getter self-heal throws a GBrainError whose `problem` field is
|
||||
// 'No database connection' — the existing typed-shape match must keep firing.
|
||||
test('matches the instance-pool-reaped GBrainError shape (problem field)', () => {
|
||||
const err = { problem: 'No database connection', message: 'instance pool torn down' };
|
||||
expect(isRetryableConnError(err)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('isRetryableError', () => {
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
/**
|
||||
* issue #1678 — cgroup-aware auto-sized RSS watchdog default.
|
||||
*
|
||||
* The load-bearing case (Codex #5): a tiny cgroup limit on a huge host must
|
||||
* win, so the watchdog cap sits BELOW the real ceiling and the graceful drain
|
||||
* beats the kernel OOM-killer. Plain os.totalmem() would pick a 16GB cap on a
|
||||
* 4GB-limited container and re-break into a silent SIGKILL.
|
||||
*/
|
||||
|
||||
import { describe, it, expect } from 'bun:test';
|
||||
import {
|
||||
resolveDefaultMaxRssMb,
|
||||
describeDefaultMaxRss,
|
||||
readCgroupMemLimitBytes,
|
||||
RSS_DEFAULT_FLOOR_MB,
|
||||
RSS_DEFAULT_CEIL_MB,
|
||||
} from '../src/core/minions/rss-default.ts';
|
||||
|
||||
const GB = 1024 * 1024 * 1024;
|
||||
|
||||
describe('resolveDefaultMaxRssMb — clamp', () => {
|
||||
it('8GB host (no cgroup) → floor 4096', () => {
|
||||
expect(resolveDefaultMaxRssMb({ totalMemBytes: 8 * GB, cgroupLimitBytes: null })).toBe(4096);
|
||||
});
|
||||
|
||||
it('16GB host → 8192 (0.5x, inside the band)', () => {
|
||||
expect(resolveDefaultMaxRssMb({ totalMemBytes: 16 * GB, cgroupLimitBytes: null })).toBe(8192);
|
||||
});
|
||||
|
||||
it('32GB host → ceil 16384', () => {
|
||||
expect(resolveDefaultMaxRssMb({ totalMemBytes: 32 * GB, cgroupLimitBytes: null })).toBe(16384);
|
||||
});
|
||||
|
||||
it('126GB host → ceil 16384 (the incident box)', () => {
|
||||
expect(resolveDefaultMaxRssMb({ totalMemBytes: 126 * GB, cgroupLimitBytes: null })).toBe(16384);
|
||||
});
|
||||
|
||||
it('result always within [floor, ceil] for huge hosts', () => {
|
||||
const mb = resolveDefaultMaxRssMb({ totalMemBytes: 1024 * GB, cgroupLimitBytes: null });
|
||||
expect(mb).toBeGreaterThanOrEqual(RSS_DEFAULT_FLOOR_MB);
|
||||
expect(mb).toBeLessThanOrEqual(RSS_DEFAULT_CEIL_MB);
|
||||
});
|
||||
});
|
||||
|
||||
describe('resolveDefaultMaxRssMb — cgroup limit wins (Codex #5)', () => {
|
||||
it('4GB cgroup on a 126GB host → cap stays BELOW the 4GB ceiling', () => {
|
||||
const d = describeDefaultMaxRss({ totalMemBytes: 126 * GB, cgroupLimitBytes: 4 * GB });
|
||||
expect(d.source).toBe('cgroup-limited');
|
||||
expect(d.basisMb).toBe(4096);
|
||||
// 0.5x4096 = 2048, below the 4096 floor but the floor must NOT push the cap
|
||||
// up to/above the real 4GB ceiling — that would defeat drain-before-OOM.
|
||||
expect(d.mb).toBeLessThan(4096);
|
||||
expect(d.mb).toBe(2048);
|
||||
});
|
||||
|
||||
it('8GB cgroup on a big host → 4096 (0.5x), source cgroup-limited', () => {
|
||||
const d = describeDefaultMaxRss({ totalMemBytes: 64 * GB, cgroupLimitBytes: 8 * GB });
|
||||
expect(d.mb).toBe(4096);
|
||||
expect(d.source).toBe('cgroup-limited');
|
||||
});
|
||||
|
||||
it('cgroup limit >= host RAM reads as host (unlimited sentinel collapses via min)', () => {
|
||||
const d = describeDefaultMaxRss({ totalMemBytes: 16 * GB, cgroupLimitBytes: 9_223_372_036_854_771_712 });
|
||||
expect(d.source).toBe('host');
|
||||
expect(d.mb).toBe(8192);
|
||||
});
|
||||
});
|
||||
|
||||
describe('readCgroupMemLimitBytes', () => {
|
||||
it('cgroup v2 "max" → null (no enforced limit)', () => {
|
||||
const read = (p: string) => {
|
||||
if (p === '/sys/fs/cgroup/memory.max') return 'max\n';
|
||||
throw new Error('ENOENT');
|
||||
};
|
||||
expect(readCgroupMemLimitBytes(read)).toBeNull();
|
||||
});
|
||||
|
||||
it('cgroup v2 numeric → that value', () => {
|
||||
const read = (p: string) => {
|
||||
if (p === '/sys/fs/cgroup/memory.max') return String(4 * GB) + '\n';
|
||||
throw new Error('ENOENT');
|
||||
};
|
||||
expect(readCgroupMemLimitBytes(read)).toBe(4 * GB);
|
||||
});
|
||||
|
||||
it('falls back to cgroup v1 when v2 unreadable', () => {
|
||||
const read = (p: string) => {
|
||||
if (p === '/sys/fs/cgroup/memory/memory.limit_in_bytes') return String(2 * GB);
|
||||
throw new Error('ENOENT');
|
||||
};
|
||||
expect(readCgroupMemLimitBytes(read)).toBe(2 * GB);
|
||||
});
|
||||
|
||||
it('neither file present → null', () => {
|
||||
const read = () => { throw new Error('ENOENT'); };
|
||||
expect(readCgroupMemLimitBytes(read)).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -130,11 +130,35 @@ describe('summarizeCrashes — aggregation', () => {
|
||||
const summary = summarizeCrashes([]);
|
||||
expect(summary).toEqual({
|
||||
total: 0,
|
||||
by_cause: { runtime_error: 0, oom_or_external_kill: 0, unknown: 0, legacy: 0 },
|
||||
by_cause: { runtime_error: 0, oom_or_external_kill: 0, rss_watchdog: 0, unknown: 0, legacy: 0 },
|
||||
clean_exits: 0,
|
||||
});
|
||||
});
|
||||
|
||||
// issue #1678: rss_watchdog is a crash-classified cause (NOT in
|
||||
// CLEAN_EXIT_CAUSES) with its OWN bucket — operators watching
|
||||
// by_cause.rss_watchdog rise know the cap is too low for the workload, a
|
||||
// distinct signal from a generic runtime_error or OOM-killer SIGKILL.
|
||||
test('rss_watchdog routes to its own bucket, not legacy', () => {
|
||||
const summary = summarizeCrashes([
|
||||
evt('worker_exited', { likely_cause: 'rss_watchdog' }),
|
||||
evt('worker_exited', { likely_cause: 'rss_watchdog' }),
|
||||
evt('worker_exited', { likely_cause: 'runtime_error' }),
|
||||
]);
|
||||
expect(summary.total).toBe(3);
|
||||
expect(summary.by_cause.rss_watchdog).toBe(2);
|
||||
expect(summary.by_cause.runtime_error).toBe(1);
|
||||
expect(summary.by_cause.legacy).toBe(0);
|
||||
expect(summary.clean_exits).toBe(0);
|
||||
});
|
||||
|
||||
// isCrashExit treats rss_watchdog as a crash (it's a real problem), NOT a
|
||||
// clean exit — pins that the worker draining itself on a too-low cap shows
|
||||
// up in operator health surfaces instead of looking like a clean drain.
|
||||
test('isCrashExit classifies rss_watchdog as a crash', () => {
|
||||
expect(isCrashExit(evt('worker_exited', { likely_cause: 'rss_watchdog' }))).toBe(true);
|
||||
});
|
||||
|
||||
test('only non-exit events returns zero summary', () => {
|
||||
const summary = summarizeCrashes([evt('started'), evt('worker_spawned'), evt('stopped')]);
|
||||
expect(summary.total).toBe(0);
|
||||
|
||||
+37
-7
@@ -414,21 +414,20 @@ describe('MinionSupervisor', () => {
|
||||
}, 15_000);
|
||||
});
|
||||
|
||||
describe('integration: --max-rss spawn args (v0.21)', () => {
|
||||
it('passes --max-rss 2048 to spawned worker by default', async () => {
|
||||
describe('integration: --max-rss spawn args (v0.21, auto-sized v0.41.39.0)', () => {
|
||||
it('passes an explicit --max-rss through to the spawned worker', async () => {
|
||||
const outFile = join(tmpdir(), `gbrain-sup-maxrss-${process.pid}-${Date.now()}.txt`);
|
||||
try { unlinkSync(outFile); } catch { /* may not exist */ }
|
||||
|
||||
// Worker logs its argv to OUT_FILE so the test can assert --max-rss 2048
|
||||
// landed there. spawnOnce in supervisor.ts builds:
|
||||
// ['jobs', 'work', '--concurrency', '1', '--queue', 'default', '--max-rss', '2048']
|
||||
// exit 1 required post-D1/D2: code=0 workers respawn forever.
|
||||
const h = makeHarness('maxrss-default', `printf '%s\\n' "$*" > "$OUT_FILE" ; exit 1`);
|
||||
// SUP_MAX_RSS pins an explicit cap; the supervisor must pass it through
|
||||
// verbatim. exit 1 required post-D1/D2: code=0 workers respawn forever.
|
||||
const h = makeHarness('maxrss-explicit', `printf '%s\\n' "$*" > "$OUT_FILE" ; exit 1`);
|
||||
|
||||
try {
|
||||
const sup = spawnSupervisor(h, {
|
||||
OUT_FILE: outFile,
|
||||
SUP_MAX_CRASHES: '1',
|
||||
SUP_MAX_RSS: '2048',
|
||||
});
|
||||
|
||||
await sup.exited;
|
||||
@@ -441,6 +440,37 @@ describe('MinionSupervisor', () => {
|
||||
h.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
// issue #1678: with no explicit cap the supervisor auto-sizes cgroup-aware
|
||||
// instead of the old flat 2048 footgun. Same machine → the in-test
|
||||
// resolveDefaultMaxRssMb() equals what the spawned supervisor computes.
|
||||
it('auto-sizes --max-rss when no explicit cap is given', async () => {
|
||||
const outFile = join(tmpdir(), `gbrain-sup-maxrss-auto-${process.pid}-${Date.now()}.txt`);
|
||||
try { unlinkSync(outFile); } catch { /* may not exist */ }
|
||||
|
||||
const { resolveDefaultMaxRssMb } = await import('../src/core/minions/rss-default.ts');
|
||||
const expected = resolveDefaultMaxRssMb();
|
||||
|
||||
const h = makeHarness('maxrss-auto', `printf '%s\\n' "$*" > "$OUT_FILE" ; exit 1`);
|
||||
try {
|
||||
const sup = spawnSupervisor(h, {
|
||||
OUT_FILE: outFile,
|
||||
SUP_MAX_CRASHES: '1',
|
||||
});
|
||||
await sup.exited;
|
||||
|
||||
expect(existsSync(outFile)).toBe(true);
|
||||
const argv = readFileSync(outFile, 'utf8').trim();
|
||||
expect(argv).toContain(`--max-rss ${expected}`);
|
||||
// Auto-sized value is clamped into the sane range, never the old 2048
|
||||
// unless the box genuinely resolves there.
|
||||
expect(expected).toBeGreaterThanOrEqual(4096);
|
||||
expect(expected).toBeLessThanOrEqual(16384);
|
||||
} finally {
|
||||
try { unlinkSync(outFile); } catch { /* noop */ }
|
||||
h.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
});
|
||||
|
||||
describe('integration: audit file rotation + helper', () => {
|
||||
|
||||
@@ -483,3 +483,69 @@ describe('resolveLockRenewalKnobs', () => {
|
||||
expect(resolveLockRenewalKnobs({}, 30_000).maxFailuresForAudit).toBe(3);
|
||||
});
|
||||
});
|
||||
|
||||
// issue #1678 (Codex #2): the bounded reconnect-once hook. NOT a withRetry on
|
||||
// renewLock (that races this tick's own timeout); a single pool rebuild before
|
||||
// the next tick so the next renewLock hits a live connection.
|
||||
describe('runLockRenewalTick: reconnect-once dep (issue #1678)', () => {
|
||||
test('reconnect fires after a renewLock throw within deadline; result still ok', async () => {
|
||||
const audit = freshAudit();
|
||||
let reconnectCalls = 0;
|
||||
const deps: LockRenewalDeps = {
|
||||
renewLock: async () => { throw new Error('write CONNECTION_ENDED'); },
|
||||
audit: audit.sink,
|
||||
now: () => 1000, // well within deadline
|
||||
setTimeout: makeFakeTimer().setTimeout,
|
||||
reconnect: async () => { reconnectCalls++; },
|
||||
};
|
||||
const state = makeState({ lastSuccessfulRenewalAt: 0 });
|
||||
const result = await runLockRenewalTick(deps, state);
|
||||
expect(result).toEqual({ kind: 'ok' });
|
||||
expect(reconnectCalls).toBe(1);
|
||||
expect(state.consecutiveFailures).toBe(1);
|
||||
});
|
||||
|
||||
test('a reconnect throw is swallowed — tick still returns ok (no unhandledRejection class)', async () => {
|
||||
const audit = freshAudit();
|
||||
const deps: LockRenewalDeps = {
|
||||
renewLock: async () => { throw new Error('Connection terminated unexpectedly'); },
|
||||
audit: audit.sink,
|
||||
now: () => 1000,
|
||||
setTimeout: makeFakeTimer().setTimeout,
|
||||
reconnect: async () => { throw new Error('reconnect failed: EHOSTUNREACH'); },
|
||||
};
|
||||
const state = makeState({ lastSuccessfulRenewalAt: 0 });
|
||||
const result = await runLockRenewalTick(deps, state);
|
||||
expect(result).toEqual({ kind: 'ok' });
|
||||
});
|
||||
|
||||
test('reconnect is NOT called on a successful renewal', async () => {
|
||||
let reconnectCalls = 0;
|
||||
const deps: LockRenewalDeps = {
|
||||
renewLock: async () => true,
|
||||
audit: freshAudit().sink,
|
||||
now: () => 1000,
|
||||
setTimeout: makeFakeTimer().setTimeout,
|
||||
reconnect: async () => { reconnectCalls++; },
|
||||
};
|
||||
const result = await runLockRenewalTick(deps, makeState({ lastSuccessfulRenewalAt: 500 }));
|
||||
expect(result).toEqual({ kind: 'ok' });
|
||||
expect(reconnectCalls).toBe(0);
|
||||
});
|
||||
|
||||
test('reconnect is NOT called when the tick aborts at the deadline', async () => {
|
||||
let reconnectCalls = 0;
|
||||
const deps: LockRenewalDeps = {
|
||||
renewLock: async () => { throw new Error('write CONNECTION_ENDED'); },
|
||||
audit: freshAudit().sink,
|
||||
// sinceLastSuccess = 30000 - 0 = 30000 >= deadline (30000-5000=25000) → abort
|
||||
now: () => 30_000,
|
||||
setTimeout: makeFakeTimer().setTimeout,
|
||||
reconnect: async () => { reconnectCalls++; },
|
||||
};
|
||||
const state = makeState({ lastSuccessfulRenewalAt: 0 });
|
||||
const result = await runLockRenewalTick(deps, state);
|
||||
expect(result).toEqual({ kind: 'should_abort', reason: 'lock-renewal-failed' });
|
||||
expect(reconnectCalls).toBe(0); // pointless to reconnect when we're giving up the lock
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
/**
|
||||
* issue #1678 — worker-side RSS watchdog behavior.
|
||||
*
|
||||
* Pins that:
|
||||
* 1. Crossing the cap sets `rssWatchdogTriggered` (the flag the CLI reads to
|
||||
* exit with WORKER_EXIT_RSS_WATCHDOG) and drains the worker.
|
||||
* 2. The 80%-of-cap soft warn fires BEFORE the kill, once per crossing,
|
||||
* carrying the peak + in-flight job kinds — and does NOT drain.
|
||||
*
|
||||
* Uses a real in-memory PGLite engine (canonical block per CLAUDE.md R3+R4)
|
||||
* + a stubbed `getRss` so the test is deterministic and hermetic.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeAll, afterAll } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { MinionWorker } from '../src/core/minions/worker.ts';
|
||||
|
||||
const MB = 1024 * 1024;
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
describe('worker RSS watchdog (issue #1678)', () => {
|
||||
it('crossing the cap sets rssWatchdogTriggered and drains', async () => {
|
||||
const worker = new MinionWorker(engine, {
|
||||
queue: 'default',
|
||||
concurrency: 1,
|
||||
maxRssMb: 100,
|
||||
getRss: () => 500 * MB, // 5x the cap
|
||||
rssCheckInterval: 25,
|
||||
healthCheckInterval: 0, // no self-health timer in this test
|
||||
pollInterval: 25,
|
||||
});
|
||||
worker.register('noop', async () => {});
|
||||
|
||||
// start() resolves on its own: the periodic check trips the watchdog,
|
||||
// gracefulShutdown sets running=false, the loop exits.
|
||||
await worker.start();
|
||||
expect(worker.rssWatchdogTriggered).toBe(true);
|
||||
});
|
||||
|
||||
it('80% soft-warn fires before the kill and does not drain', async () => {
|
||||
const warns: string[] = [];
|
||||
const origWarn = console.warn;
|
||||
console.warn = (...a: unknown[]) => { warns.push(a.join(' ')); };
|
||||
|
||||
const worker = new MinionWorker(engine, {
|
||||
queue: 'default',
|
||||
concurrency: 1,
|
||||
maxRssMb: 100,
|
||||
getRss: () => 85 * MB, // 85% — above soft line, below cap
|
||||
rssCheckInterval: 25,
|
||||
healthCheckInterval: 0,
|
||||
pollInterval: 25,
|
||||
});
|
||||
worker.register('noop', async () => {});
|
||||
|
||||
const runPromise = worker.start();
|
||||
// Let a couple of periodic checks fire.
|
||||
await new Promise((r) => setTimeout(r, 120));
|
||||
worker.stop();
|
||||
await runPromise;
|
||||
console.warn = origWarn;
|
||||
|
||||
const softWarn = warns.find((w) => w.includes('approaching cap'));
|
||||
expect(softWarn).toBeDefined();
|
||||
expect(softWarn).toContain('85%');
|
||||
// Soft warn must NOT have drained the worker.
|
||||
expect(worker.rssWatchdogTriggered).toBe(false);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user