diff --git a/CHANGELOG.md b/CHANGELOG.md index cb8d65bd4..ec8984eb8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,84 @@ All notable changes to GBrain will be documented in this file. +## [0.22.2] - 2026-04-26 + +**Worker no longer freezes silently. Restart-on-RSS, cold-start retry, autopilot backpressure.** + +The minions worker has been freezing every few hours in production. RSS climbs from 68 MB at boot to ~15 GB over ~7 hours, the process stops claiming jobs but never crashes (no OOM, no SIGSEGV), the cron keeps enqueuing autopilot-cycle jobs every 5 minutes into a queue nobody is draining, and within 2-3 hours the queue piles up to 28+ waiting jobs. Shell jobs in flight when the worker froze hit `max_stalled` and dead-letter, producing an 18% shell-job failure rate over 24h. The brainstorm caught the root chain ... memory leak, wedged worker, supervisor cold-start race, no backpressure ... and v0.22.2 ships the three in-repo defenses that close the cascade end-to-end while the underlying memory leak gets investigated separately. + +The watchdog is the keystone. The worker now self-terminates when RSS crosses a threshold (default 2048 MB under the supervisor) and the supervisor's exponential-backoff respawn picks up a fresh process. Both per-job AND a 60-second periodic timer check, so the watchdog still fires when every concurrency slot is wedged and zero jobs are completing ... the actual production freeze pattern. On trip, the worker fires `shutdownAbort` (so the shell handler runs its SIGTERM→5s→SIGKILL cleanup on child processes) and aborts every per-job signal (so cooperative handlers bail instead of waiting out the 30s drain). Closes the zombie-shell-children gap a Codex review surfaced. + +Cold-start auth races on container boot are gone. Every CLI command's `connectEngine()` bootstrap retries transient errors (3 attempts, 1s/2s/4s backoff) by default. PgBouncer rejecting the first connect on a freshly-pinged Supabase pooler is the production failure mode that killed autopilot on cold start; the retry handles it transparently. Operators who genuinely want fail-fast on a misconfigured `DATABASE_URL` pass `--no-retry-connect` or set `GBRAIN_NO_RETRY_CONNECT=1`. + +Autopilot stops piling jobs into a dead queue. `autopilot-cycle` submissions now use `maxWaiting: 1` so the v0.19.1 `pg_advisory_xact_lock` coalesce path caps the queue at 1 active + 1 waiting instead of letting it grow unbounded. The 3rd+ submission coalesces and writes a backpressure-audit JSONL line. Combined with the existing per-slot `idempotency_key`, cross-slot pile-ups are bounded. + +### The numbers that matter + +Production data from the 2026-04-25 incident, plus the watchdog defaults: + +| Metric | Before | After (supervised path) | +|-----------------------------------------|-----------------|-------------------------| +| Waiting-jobs pileup at freeze | 28+ | 2 (capped at 1+1) | +| Worker RSS at freeze | 14.8 GB | ~2 GB self-terminate | +| Time to detect freeze | hours (manual) | ≤60s (periodic timer) | +| Cold-start auth-fail recovery | manual restart | 3 attempts in ~7s | + +Bare `gbrain jobs work` (operators not using the supervisor) keeps current unbounded behavior to preserve workloads with legitimately large embed/import working sets ... pass `--max-rss N` explicitly to enable the watchdog there. + +### What this means for operators + +If you run `gbrain jobs supervisor` (the production-recommended path), `gbrain upgrade` is the only step. The supervisor injects `--max-rss 2048` to its spawned worker by default; hourly watchdog exits look like clean shutdowns to the supervisor's stable-run reset, not crashes. If you run `gbrain autopilot --install`, the autopilot's worker spawn loop now has the same stable-run reset pattern, so a watchdog-driven exit every hour does NOT trip the give-up-after-5-crashes threshold. If your container hits zombie process accumulation, add `--init` to `docker run` or `tini` as PID 1 ... that's a host-side concern, not a gbrain change. + +## To take advantage of v0.22.2 + +`gbrain upgrade` should do this automatically. If it didn't, or if `gbrain doctor` warns about a partial migration: + +1. **Run the orchestrator manually:** + ```bash + gbrain apply-migrations --yes + ``` +2. **No manual SKILL.md or AGENTS.md edits required.** This release is code-only ... no schema changes, no new skills. +3. **Verify the watchdog is wired (Postgres + supervisor path):** + ```bash + gbrain jobs supervisor --json & + ps -ef | grep "gbrain jobs work" | grep -- "--max-rss 2048" + ``` + You should see the spawned worker child carrying `--max-rss 2048` in its argv. +4. **If you supervise via `gbrain autopilot --install`,** the watchdog gets injected automatically. Existing crontab/launchd/systemd installs do not need to be reinstalled ... the autopilot binary picks up the new spawn args on next restart. +5. **For hosts hitting zombie process accumulation** (PID-table fills up over weeks): add `--init` to `docker run`, or set `tini` as PID 1 in your Dockerfile. Not a gbrain code change ... operational note. +6. **If any step fails or behavior looks off,** please file an issue at https://github.com/garrytan/gbrain/issues with the output of `gbrain doctor` and the contents of `~/.gbrain/upgrade-errors.jsonl` if it exists. + +### Itemized changes + +#### Added + +- `MinionWorkerOpts` gains `maxRssMb`, `getRss`, and `rssCheckInterval` ... watchdog plumbing with a deterministic-test seam for the RSS readback. +- `MinionWorker.gracefulShutdown(reason)` ... unified-style shutdown that fires `shutdownAbort` + per-job aborts + `running=false`. Reused by the per-job and periodic-timer check sites. +- 60-second periodic RSS check (`rssCheckInterval` default 60_000) running alongside the existing stalled-jobs timer in `start()`. Closes the freeze-with-zero-completions production scenario. +- `--max-rss MB` flag on `gbrain jobs work` (no default, opt-in for bare workers) and `gbrain jobs supervisor` (default 2048). `--max-rss 0` disables; `< 256` errors out as a likely GB-vs-MB unit-confusion typo. +- `connectWithRetry()` + `isRetryableDbConnectError()` in `src/core/db.ts`. 5-pattern transient-error matcher (auth-failed, connection-refused, db-starting, terminated-unexpectedly, ECONNRESET). Permanent errors (extension-missing, schema conflicts) do NOT retry. +- `--no-retry-connect` flag and `GBRAIN_NO_RETRY_CONNECT=1` env var ... operator escape hatch for fail-fast on misconfigured DATABASE_URL. +- Autopilot worker spawn now carries `--max-rss 2048` and a stable-run reset window (5 minutes uptime → reset crash counter to 1). Mirrors the supervisor pattern at `supervisor.ts:471-476` so hourly watchdog exits don't kill autopilot after ~5 hours. +- `autopilot-cycle` submission passes `maxWaiting: 1` to `queue.add()`. Combined with the existing per-slot `idempotency_key`, this caps cross-slot queue depth at 1 active + 1 waiting. +- 11 new tests in `test/minions.test.ts` covering the watchdog (5 cases including the production-freeze-regression case where zero jobs ever complete) and `connectWithRetry` (6 cases including the noRetry opt-out, transient/permanent error distinction, and successful retry). +- New supervisor integration test asserting `--max-rss 2048` lands in the spawned worker's argv by default. + +#### Changed + +- `MinionSupervisor` `SupervisorOpts` gains `maxRssMb` (default 2048). The spawn-args builder appends `--max-rss N` when `maxRssMb > 0`. +- `connectEngine()` in `src/cli.ts` now wraps `engine.connect()` in `connectWithRetry` by default. Behavior change for cold-start auth races; preserve original fail-fast with `--no-retry-connect` per call site. + +#### Out of scope (follow-ups) + +- The 40 MB/job memory leak itself ... separate investigation needs heap snapshots and a real reproducer. The watchdog removes urgency. +- Zombie process reaping via `tini` or `--init` ... Render/Docker host-side configuration, documented above. +- Refactoring SIGTERM/SIGINT/watchdog into one `unifiedShutdown(reason)` helper ... right shape long-term, premature for this PR. + +### For contributors + +- The watchdog cleanup path (`gracefulShutdown`) is intentionally co-located with `MinionWorker.stop()`. When a third caller appears (e.g., a future `pause()` method), extracting `unifiedShutdown(reason)` becomes worth the refactor. Until then, three lines is not a DRY emergency. +- `isRetryableDbConnectError()` lives in `src/core/db.ts` and owns its own 5-pattern matcher. PR #406 (when it merges) introduces a 13-pattern matcher in `src/core/minions/supervisor.ts`; the right move at that merge is to delete the supervisor's local copy and import from `db.ts` (correct dependency direction, low → high). A follow-up TODO captures this. ## [0.22.1] - 2026-04-26 **Autopilot stops being a noisy neighbor.** diff --git a/VERSION b/VERSION index a723ece79..faa5fb265 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.22.1 +0.22.2 diff --git a/package.json b/package.json index eea700a78..c293b1e2c 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "gbrain", - "version": "0.22.1", + "version": "0.22.2", "description": "Postgres-native personal knowledge brain with hybrid RAG search", "type": "module", "main": "src/core/index.ts", diff --git a/src/cli.ts b/src/cli.ts index 34fdc5d88..78da9120d 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -576,7 +576,10 @@ async function connectEngine(): Promise { } const { createEngine } = await import('./core/engine-factory.ts'); const engine = await createEngine(toEngineConfig(config)); - await engine.connect(toEngineConfig(config)); + const noRetry = process.argv.includes('--no-retry-connect') || + process.env.GBRAIN_NO_RETRY_CONNECT === '1'; + const { connectWithRetry } = await import('./core/db.ts'); + await connectWithRetry(engine, toEngineConfig(config), { noRetry }); return engine; } diff --git a/src/commands/autopilot.ts b/src/commands/autopilot.ts index 3663e4f5d..7d6b11cb9 100644 --- a/src/commands/autopilot.ts +++ b/src/commands/autopilot.ts @@ -147,22 +147,41 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { let stopping = false; let workerProc: ChildProcess | null = null; let crashCount = 0; + let lastWorkerStartTime = 0; + + // Stable-run reset window (matches MinionSupervisor.ts:471-476 pattern). If the + // worker ran > 5min before exit, treat as a fresh cycle (crashCount=1) so the + // RSS watchdog firing hourly does NOT trip autopilot's give-up threshold after + // ~5 hours of healthy uptime. + const STABLE_RUN_RESET_MS = 5 * 60 * 1000; if (spawnManagedWorker) { const cliPath = resolveGbrainCliPath(); const startWorker = () => { - const child = spawn(cliPath, ['jobs', 'work'], { stdio: 'inherit', env: process.env }); + // Inject the RSS watchdog default (2048 MB) for the autopilot-supervised + // worker. Bare `gbrain jobs work` has no default; the supervisor and + // autopilot are the production paths that opt in. + const args = ['jobs', 'work', '--max-rss', '2048']; + const child = spawn(cliPath, args, { stdio: 'inherit', env: process.env }); workerProc = child; - console.log(`[autopilot] Minions worker spawned (pid: ${child.pid})`); + lastWorkerStartTime = Date.now(); + console.log(`[autopilot] Minions worker spawned (pid: ${child.pid}, watchdog: 2048MB)`); child.on('exit', (code) => { workerProc = null; if (stopping) return; + const runDuration = Date.now() - lastWorkerStartTime; + if (runDuration > STABLE_RUN_RESET_MS) { + // Stable run — forgive prior crash history. A watchdog-driven hourly + // exit (the production path post-fix) lands here every time. + crashCount = 1; + } else { + crashCount++; + } if (crashCount >= 5) { - console.error('[autopilot] 5 consecutive worker crashes, giving up.'); + console.error(`[autopilot] 5 consecutive worker crashes (run ${runDuration}ms), giving up.`); process.exit(1); } - crashCount++; - console.error(`[autopilot] worker exited code=${code}, restart #${crashCount} in 10s`); + console.error(`[autopilot] worker exited code=${code} after ${runDuration}ms, restart #${crashCount} in 10s`); setTimeout(startWorker, 10_000); }); }; @@ -290,6 +309,12 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { idempotency_key: `autopilot-cycle:${slot}`, max_attempts: 2, timeout_ms: timeoutMs, + // Submission backpressure: when the worker is dead or wedged, + // idempotency_key only dedupes within a slot; cross-slot pile-up + // is what produced the 28+ waiting-jobs production incident. + // maxWaiting: 1 caps at 1 active + 1 waiting; queue.add coalesces + // the 3rd+ submission and writes a backpressure-audit JSONL line. + maxWaiting: 1, }, ); if (jsonMode) { diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 8ef04ddba..7e8d97479 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -32,6 +32,32 @@ export function parseMaxWaitingFlag(args: string[]): number | undefined { return Math.max(1, Math.min(100, parsed)); } +/** Parse `--max-rss N` (MB). Returns: + * - 0 if the flag is absent (no watchdog by default for bare `jobs work`) + * - 0 if `--max-rss 0` (explicit disable) + * - the value if >= 256 + * Errors and exits the process if the flag is non-numeric, negative, or + * positive but < 256 (likely a GB-vs-MB unit-confusion typo). */ +export function parseMaxRssFlag(args: string[]): number { + const raw = parseFlag(args, '--max-rss'); + if (raw === undefined) return 0; + const parsed = parseInt(raw, 10); + if (!Number.isFinite(parsed) || parsed < 0) { + console.error(`Error: --max-rss must be a non-negative integer (MB), got "${raw}"`); + process.exit(1); + } + if (parsed === 0) return 0; + if (parsed < 256) { + console.error( + `Error: --max-rss ${parsed} is too low for production (likely a unit confusion: ` + + `--max-rss takes megabytes, not gigabytes). Use --max-rss 0 to disable, ` + + `or set a value >= 256.` + ); + process.exit(1); + } + return parsed; +} + export function resolveWorkerConcurrency(args: string[], env: NodeJS.ProcessEnv = process.env): number { const raw = parseFlag(args, '--concurrency') ?? env.GBRAIN_WORKER_CONCURRENCY ?? '1'; const parsed = parseInt(raw, 10); @@ -106,11 +132,12 @@ USAGE gbrain jobs delete gbrain jobs stats gbrain jobs smoke - gbrain jobs work [--queue Q] [--concurrency N] + gbrain jobs work [--queue Q] [--concurrency N] [--max-rss MB] gbrain jobs supervisor [start] [--detach] [--json] [--concurrency N] [--queue Q] [--pid-file PATH] [--max-crashes N] [--health-interval N] [--allow-shell-jobs] [--cli-path PATH] + [--max-rss MB] gbrain jobs supervisor status [--json] [--pid-file PATH] gbrain jobs supervisor stop [--json] [--pid-file PATH] @@ -611,14 +638,19 @@ HANDLER TYPES (built in) const queueName = parseFlag(args, '--queue') ?? 'default'; const concurrency = resolveWorkerConcurrency(args); + // --max-rss is opt-in for bare `gbrain jobs work` — preserves pre-v0.21 behavior + // for operators with legitimately large embed/import working sets. The supervisor + // path injects a default 2048; this code path does not. + const maxRssMb = parseMaxRssFlag(args); try { await queue.ensureSchema(); } catch (e) { console.error(e instanceof Error ? e.message : String(e)); process.exit(1); } - const worker = new MinionWorker(engine, { queue: queueName, concurrency }); + const worker = new MinionWorker(engine, { queue: queueName, concurrency, maxRssMb }); await registerBuiltinHandlers(worker, engine); - console.log(`Minion worker started (queue: ${queueName}, concurrency: ${concurrency})`); + const watchdogNote = maxRssMb > 0 ? `, watchdog: ${maxRssMb}MB` : ''; + console.log(`Minion worker started (queue: ${queueName}, concurrency: ${concurrency}${watchdogNote})`); console.log(`Registered handlers: ${worker.registeredNames.join(', ')}`); await worker.start(); break; @@ -759,6 +791,11 @@ HANDLER TYPES (built in) const allowShellJobs = hasFlag(args, '--allow-shell-jobs') || !!process.env.GBRAIN_ALLOW_SHELL_JOBS; const detach = hasFlag(args, '--detach'); + // Supervisor defaults --max-rss 2048 (MB) — main production path uses + // the supervisor, so the watchdog is on by default here. parseMaxRssFlag + // returns 0 when the flag is absent; substitute the supervisor default. + const maxRssRaw = parseMaxRssFlag(args); + const maxRssMb = parseFlag(args, '--max-rss') === undefined ? 2048 : maxRssRaw; const cliPath = parseFlag(args, '--cli-path') ?? resolveGbrainCliPath(); @@ -796,6 +833,7 @@ HANDLER TYPES (built in) cliPath, allowShellJobs, json: jsonMode, + maxRssMb, onEvent: (emission) => writeSupervisorEvent(emission, supervisorPid), }); diff --git a/src/core/db.ts b/src/core/db.ts index 6093d3878..10f18175e 100644 --- a/src/core/db.ts +++ b/src/core/db.ts @@ -1,6 +1,7 @@ import postgres from 'postgres'; import { GBrainError, type EngineConfig } from './types.ts'; import { SCHEMA_SQL } from './schema-embedded.ts'; +import type { BrainEngine } from './engine.ts'; let sql: ReturnType | null = null; let connectedUrl: string | null = null; @@ -242,3 +243,56 @@ export async function withTransaction(fn: (tx: ReturnType) = return fn(tx as unknown as ReturnType); }) as Promise; } + +const RETRYABLE_DB_CONNECT_PATTERNS = [ + /password authentication failed/i, + /connection refused/i, + /the database system is starting up/i, + /Connection terminated unexpectedly/i, + /ECONNRESET/i, +]; + +export function isRetryableDbConnectError(err: unknown): boolean { + const msg = err instanceof Error ? err.message : String(err); + if (!msg) return false; + return RETRYABLE_DB_CONNECT_PATTERNS.some(p => p.test(msg)); +} + +export interface ConnectWithRetryOpts { + attempts?: number; + baseDelayMs?: number; + noRetry?: boolean; + log?: (line: string) => void; +} + +export async function connectWithRetry( + engine: BrainEngine, + config: EngineConfig & { poolSize?: number }, + opts: ConnectWithRetryOpts = {}, +): Promise { + const noRetry = opts.noRetry ?? (process.env.GBRAIN_NO_RETRY_CONNECT === '1'); + const attempts = noRetry ? 1 : (opts.attempts ?? 3); + const baseDelayMs = opts.baseDelayMs ?? 1000; + const log = opts.log ?? ((line) => console.warn(line)); + + let lastErr: unknown; + for (let i = 0; i < attempts; i++) { + try { + await engine.connect(config); + return; + } catch (e: unknown) { + lastErr = e; + const retryable = isRetryableDbConnectError(e); + const isLast = i === attempts - 1; + if (!retryable || isLast) { + throw e; + } + const delay = baseDelayMs * Math.pow(2, i); + const msg = e instanceof Error ? e.message : String(e); + log(`[connect] attempt ${i + 1} failed (${msg.slice(0, 80)}), retrying in ${delay}ms`); + await new Promise(resolve => setTimeout(resolve, delay)); + } + } + // Unreachable, but TS needs the throw. + throw lastErr; +} diff --git a/src/core/minions/supervisor.ts b/src/core/minions/supervisor.ts index 8de710d25..c3f70401d 100644 --- a/src/core/minions/supervisor.ts +++ b/src/core/minions/supervisor.ts @@ -75,6 +75,9 @@ export interface SupervisorOpts { allowShellJobs: boolean; /** JSON mode: emit JSONL events on stderr, reserve stdout for data payloads. Default: false. */ json: boolean; + /** RSS threshold (MB) passed to the spawned worker as `--max-rss N`. + * Default: 2048. Set to 0 to spawn the worker without a watchdog. */ + maxRssMb: number; /** Optional event sink (Lane C audit writer). Called for every lifecycle event. */ onEvent?: (event: SupervisorEmission) => void; /** @@ -101,6 +104,7 @@ const DEFAULTS: Omit = { healthInterval: 60_000, allowShellJobs: false, json: false, + maxRssMb: 2048, }; /** Calculate backoff: 1s, 2s, 4s, 8s, 16s, 32s, 60s cap. */ @@ -411,6 +415,9 @@ export class MinionSupervisor { '--concurrency', String(this.opts.concurrency), '--queue', this.opts.queue, ]; + if (this.opts.maxRssMb > 0) { + args.push('--max-rss', String(this.opts.maxRssMb)); + } // Build child env. Explicit handling for GBRAIN_ALLOW_SHELL_JOBS: // inherit only when caller opts in, otherwise strip from the clone. diff --git a/src/core/minions/types.ts b/src/core/minions/types.ts index fc7ca5e7e..912f50b67 100644 --- a/src/core/minions/types.ts +++ b/src/core/minions/types.ts @@ -160,6 +160,16 @@ export interface MinionWorkerOpts { stalledInterval?: number; // ms, default 30000 maxStalledCount?: number; // default 1 pollInterval?: number; // ms, default 5000 (for PGLite fallback) + /** RSS threshold in MB. When exceeded, worker triggers graceful shutdown + * so a supervisor can respawn it. 0 or undefined = disabled. */ + maxRssMb?: number; + /** Optional injection point for RSS readback. Defaults to + * `() => process.memoryUsage().rss`. Tests inject deterministic sequences. */ + getRss?: () => number; + /** Periodic RSS check interval in ms, default 60000. Catches the freeze + * case where all concurrency slots are wedged with zero job completions + * so the per-job check never fires. */ + rssCheckInterval?: number; } // --- Job Context (passed to handlers) --- diff --git a/src/core/minions/worker.ts b/src/core/minions/worker.ts index 6db549282..87bc1b8af 100644 --- a/src/core/minions/worker.ts +++ b/src/core/minions/worker.ts @@ -56,6 +56,11 @@ export class MinionWorker { * deploy restart — they still get the full 30s cleanup race instead. */ private shutdownAbort = new AbortController(); + /** Cumulative jobs that finished (success or failure). Used in watchdog log lines. */ + private jobsCompleted = 0; + /** Idempotency latch for gracefulShutdown — per-job and periodic check sites can race. */ + private gracefulShutdownFired = false; + private opts: Required; constructor( @@ -73,6 +78,9 @@ export class MinionWorker { stalledInterval: opts?.stalledInterval ?? 30000, maxStalledCount: opts?.maxStalledCount ?? 1, pollInterval: opts?.pollInterval ?? 5000, + maxRssMb: opts?.maxRssMb ?? 0, + getRss: opts?.getRss ?? (() => process.memoryUsage().rss), + rssCheckInterval: opts?.rssCheckInterval ?? 60000, }; } @@ -136,6 +144,17 @@ export class MinionWorker { } }, this.opts.stalledInterval); + // Periodic RSS watchdog — closes the production-freeze regression where + // all concurrency slots are wedged with zero job completions, so the + // per-job check in executeJob().finally() never fires. Disabled when + // maxRssMb is 0 (default for bare `gbrain jobs work`; supervisor sets 2048). + let rssTimer: ReturnType | null = null; + if (this.opts.maxRssMb > 0) { + rssTimer = setInterval(() => { + this.checkMemoryLimit('periodic'); + }, this.opts.rssCheckInterval); + } + try { while (this.running) { // Promote delayed jobs @@ -181,6 +200,7 @@ export class MinionWorker { } } finally { clearInterval(stalledTimer); + if (rssTimer) clearInterval(rssTimer); process.removeListener('SIGTERM', shutdown); process.removeListener('SIGINT', shutdown); @@ -257,6 +277,55 @@ export class MinionWorker { this.running = false; } + /** RSS watchdog. Called from the per-job finally and the periodic timer. + * Idempotent: returns early if already not running or already shut down. + * When threshold is exceeded, hands off to gracefulShutdown(). */ + private checkMemoryLimit(source: 'post-job' | 'periodic'): void { + if (this.opts.maxRssMb <= 0) return; + if (!this.running) return; + if (this.gracefulShutdownFired) return; + + let rss = 0; + try { + rss = this.opts.getRss(); + } catch { + // process.memoryUsage() effectively cannot throw, but be safe. + return; + } + const rssMb = Math.round(rss / (1024 * 1024)); + if (rssMb < this.opts.maxRssMb) return; + + const ts = new Date().toISOString().slice(11, 19); + console.warn( + `[watchdog ${ts}] rss=${rssMb}MB threshold=${this.opts.maxRssMb}MB ` + + `jobs_completed=${this.jobsCompleted} source=${source} — draining`, + ); + this.gracefulShutdown('watchdog'); + } + + /** Trigger a unified-style graceful shutdown. Fires shutdownAbort + per-job + * aborts + running=false in that order so: + * 1. Shell handlers (and anything subscribed to ctx.shutdownSignal) start + * their cleanup sequence (SIGTERM → 5s grace → SIGKILL on children). + * 2. Cooperative handlers see ctx.signal.aborted and bail instead of + * waiting out the 30s drain. + * 3. Main loop exits at the top of the next iteration. + * The existing 30s drain in start()'s finally then backstops genuinely + * uninterruptible work. */ + private gracefulShutdown(reason: string): void { + if (this.gracefulShutdownFired) return; + this.gracefulShutdownFired = true; + if (!this.shutdownAbort.signal.aborted) { + this.shutdownAbort.abort(new Error(reason)); + } + for (const entry of this.inFlight.values()) { + if (!entry.abort.signal.aborted) { + entry.abort.abort(new Error(reason)); + } + } + this.running = false; + } + /** Launch a job as an independent in-flight promise. */ private launchJob(job: MinionJob, lockToken: string): void { const abort = new AbortController(); @@ -310,6 +379,8 @@ export class MinionWorker { if (timeoutTimer) clearTimeout(timeoutTimer); if (graceTimer) clearTimeout(graceTimer); this.inFlight.delete(job.id); + this.jobsCompleted += 1; + this.checkMemoryLimit('post-job'); }); this.inFlight.set(job.id, { job, lockToken, lockTimer, abort, promise }); diff --git a/test/minions.test.ts b/test/minions.test.ts index 68e87b675..dadec61ea 100644 --- a/test/minions.test.ts +++ b/test/minions.test.ts @@ -1897,6 +1897,261 @@ describe('MinionQueue: v0.19.1 wall-clock + handleTimeouts non-interference (T1) }); }); +// --- v0.22.2: RSS watchdog (--max-rss + periodic timer + gracefulShutdown) --- + +describe('MinionWorker: --max-rss watchdog', () => { + // Helper: build a worker with deterministic RSS injection. Tests pass a + // sequence of bytes; getRss() returns elements in order, repeating the last. + function makeRssSequence(values: number[]): () => number { + let i = 0; + return () => { + const v = values[Math.min(i, values.length - 1)]; + i++; + return v; + }; + } + + test('per-job check: handler bumps RSS, post-job check trips, sibling aborts', async () => { + // 100MB threshold. RSS reads always return 250MB → first post-job check + // (after the 'quick' handler completes) trips and the 'slow' sibling + // sees its abort signal flip. + const worker = new MinionWorker(engine, { + queue: 'default', + concurrency: 2, + maxRssMb: 100, + getRss: () => 250 * 1024 * 1024, + pollInterval: 50, + stalledInterval: 10_000, + rssCheckInterval: 60_000, // disable periodic — exercise per-job only + }); + + let sibling2Aborted = false; + let sibling2Resolved = false; + let sibling1Done = false; + + worker.register('quick', async () => { + // Resolves immediately. Triggers post-job check. + sibling1Done = true; + }); + worker.register('slow', async (job) => { + // Long-running sibling. Watch for abort signal. + job.signal.addEventListener('abort', () => { sibling2Aborted = true; }); + await new Promise((resolve) => { + const t = setInterval(() => { + if (job.signal.aborted) { clearInterval(t); sibling2Resolved = true; resolve(); } + }, 20); + }); + }); + + await queue.add('slow', {}); + await queue.add('quick', {}); + + await worker.start(); // returns when stop() flips and drain completes + + expect(sibling1Done).toBe(true); + expect(sibling2Aborted).toBe(true); + expect(sibling2Resolved).toBe(true); + }, 60_000); + + test('periodic timer: zero job completions, watchdog still fires', async () => { + // Threshold 100MB. RSS = 250MB on every call. No job ever completes. + const worker = new MinionWorker(engine, { + queue: 'default', + concurrency: 1, + maxRssMb: 100, + getRss: () => 250 * 1024 * 1024, + pollInterval: 50, + stalledInterval: 10_000, + rssCheckInterval: 100, // fire fast in tests + }); + + let abortedDuringHandler = false; + + worker.register('forever', async (job) => { + // Never returns naturally. Wait on abort. + await new Promise((resolve) => { + const t = setInterval(() => { + if (job.signal.aborted) { + abortedDuringHandler = true; + clearInterval(t); + resolve(); + } + }, 20); + }); + }); + + await queue.add('forever', {}); + + await worker.start(); + + expect(abortedDuringHandler).toBe(true); + }, 60_000); + + test('shutdownAbort fires (closes shell-handler zombie gap)', async () => { + const worker = new MinionWorker(engine, { + queue: 'default', + concurrency: 1, + maxRssMb: 100, + getRss: () => 250 * 1024 * 1024, + pollInterval: 50, + stalledInterval: 10_000, + rssCheckInterval: 100, + }); + + let shutdownSignalFired = false; + + worker.register('observer', async (job) => { + // Subscribes to shutdownSignal — same pattern as shell.ts + job.shutdownSignal.addEventListener('abort', () => { shutdownSignalFired = true; }); + await new Promise((resolve) => { + const t = setInterval(() => { + if (job.signal.aborted) { clearInterval(t); resolve(); } + }, 20); + }); + }); + + await queue.add('observer', {}); + await worker.start(); + + expect(shutdownSignalFired).toBe(true); + }, 60_000); + + test('below threshold: no-op (no shutdown)', async () => { + let postJobCount = 0; + const worker = new MinionWorker(engine, { + queue: 'default', + concurrency: 1, + maxRssMb: 1024, + getRss: () => { postJobCount++; return 50 * 1024 * 1024; }, // always 50MB, way under + pollInterval: 50, + stalledInterval: 10_000, + rssCheckInterval: 60_000, + }); + + worker.register('noop', async () => {}); + + await queue.add('noop', {}); + await queue.add('noop', {}); + await queue.add('noop', {}); + + // Run for a moment, then stop manually + const startPromise = worker.start(); + await new Promise(r => setTimeout(r, 500)); + worker.stop(); + await startPromise; + + // Watchdog never tripped → all 3 jobs completed + const completed = await queue.getJobs({ status: 'completed' }); + expect(completed.length).toBe(3); + expect(postJobCount).toBeGreaterThanOrEqual(3); // checkMemoryLimit ran each time + }, 60_000); + + test('maxRssMb=0 disables watchdog entirely', async () => { + const worker = new MinionWorker(engine, { + queue: 'default', + concurrency: 1, + maxRssMb: 0, + getRss: () => 999_999 * 1024 * 1024, // huge, but disabled + pollInterval: 50, + stalledInterval: 10_000, + rssCheckInterval: 100, + }); + + worker.register('noop', async () => {}); + await queue.add('noop', {}); + + const startPromise = worker.start(); + await new Promise(r => setTimeout(r, 500)); + worker.stop(); + await startPromise; + + const completed = await queue.getJobs({ status: 'completed' }); + expect(completed.length).toBe(1); + }, 60_000); +}); + +// --- v0.21: connectWithRetry + isRetryableDbConnectError --- + +describe('connectWithRetry / isRetryableDbConnectError', () => { + test('isRetryableDbConnectError matches transient patterns', async () => { + const { isRetryableDbConnectError } = await import('../src/core/db.ts'); + expect(isRetryableDbConnectError(new Error('password authentication failed for user postgres'))).toBe(true); + expect(isRetryableDbConnectError(new Error('connection refused'))).toBe(true); + expect(isRetryableDbConnectError(new Error('the database system is starting up'))).toBe(true); + expect(isRetryableDbConnectError(new Error('Connection terminated unexpectedly'))).toBe(true); + expect(isRetryableDbConnectError(new Error('something happened: ECONNRESET'))).toBe(true); + }); + + test('isRetryableDbConnectError rejects permanent errors', async () => { + const { isRetryableDbConnectError } = await import('../src/core/db.ts'); + expect(isRetryableDbConnectError(new Error('extension "vector" does not exist'))).toBe(false); + expect(isRetryableDbConnectError(new Error('relation "pages" does not exist'))).toBe(false); + expect(isRetryableDbConnectError(new Error('syntax error at end of input'))).toBe(false); + }); + + test('connectWithRetry: 1st rejects transient, 2nd succeeds', async () => { + const { connectWithRetry } = await import('../src/core/db.ts'); + let attempts = 0; + const fakeEngine = { + connect: async () => { + attempts++; + if (attempts === 1) throw new Error('password authentication failed for user postgres'); + }, + } as unknown as Parameters[0]; + + await connectWithRetry(fakeEngine, { database_url: 'postgres://x' }, { baseDelayMs: 1, log: () => {} }); + expect(attempts).toBe(2); + }); + + test('connectWithRetry: 3 transient rejects → throws', async () => { + const { connectWithRetry } = await import('../src/core/db.ts'); + let attempts = 0; + const fakeEngine = { + connect: async () => { + attempts++; + throw new Error('connection refused'); + }, + } as unknown as Parameters[0]; + + await expect( + connectWithRetry(fakeEngine, { database_url: 'postgres://x' }, { baseDelayMs: 1, log: () => {} }) + ).rejects.toThrow('connection refused'); + expect(attempts).toBe(3); + }); + + test('connectWithRetry: permanent error does NOT retry', async () => { + const { connectWithRetry } = await import('../src/core/db.ts'); + let attempts = 0; + const fakeEngine = { + connect: async () => { + attempts++; + throw new Error('extension "vector" does not exist'); + }, + } as unknown as Parameters[0]; + + await expect( + connectWithRetry(fakeEngine, { database_url: 'postgres://x' }, { baseDelayMs: 1, log: () => {} }) + ).rejects.toThrow('extension "vector"'); + expect(attempts).toBe(1); + }); + + test('connectWithRetry: noRetry honored', async () => { + const { connectWithRetry } = await import('../src/core/db.ts'); + let attempts = 0; + const fakeEngine = { + connect: async () => { + attempts++; + throw new Error('connection refused'); + }, + } as unknown as Parameters[0]; + + await expect( + connectWithRetry(fakeEngine, { database_url: 'postgres://x' }, { noRetry: true, log: () => {} }) + ).rejects.toThrow(); + expect(attempts).toBe(1); + }); +}); + // --------------------------------------------------------------------------- // Abort signal propagation + force-eviction (v0.20.5 cycle-abort fix) // --------------------------------------------------------------------------- diff --git a/test/supervisor.test.ts b/test/supervisor.test.ts index 272b5ed5c..e38a6f0c3 100644 --- a/test/supervisor.test.ts +++ b/test/supervisor.test.ts @@ -328,6 +328,34 @@ describe('MinionSupervisor', () => { }, 15_000); }); + describe('integration: --max-rss spawn args (v0.21)', () => { + it('passes --max-rss 2048 to spawned worker by default', 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'] + const h = makeHarness('maxrss-default', `printf '%s\\n' "$*" > "$OUT_FILE" ; exit 0`); + + 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 2048'); + } finally { + try { unlinkSync(outFile); } catch { /* noop */ } + h.cleanup(); + } + }, 15_000); + }); + describe('integration: audit file rotation + helper', () => { it('computeSupervisorAuditFilename returns supervisor-YYYY-Www.jsonl format', () => { const jan15_2026 = new Date(Date.UTC(2026, 0, 15)); // Thu