From 16eb8cd06cff48dafb05bd11a5d5de1c7acdada8 Mon Sep 17 00:00:00 2001 From: Javier Aldape Date: Thu, 23 Jul 2026 12:24:30 -0600 Subject: [PATCH] fix(doctor): flag embed backfills without a worker (#2696) Co-authored-by: gbrain-contrib --- src/commands/doctor.ts | 351 +++++++++++++++---------------- test/doctor-wedged-queue.test.ts | 36 +++- 2 files changed, 206 insertions(+), 181 deletions(-) diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index 73dc89721..cbb1fee8a 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -757,31 +757,7 @@ export async function doctorReportRemote(engine: BrainEngine): Promise( - `SELECT COUNT(*) AS stalled FROM minion_jobs - WHERE status = 'active' - AND started_at IS NOT NULL - AND started_at < NOW() - INTERVAL '1 hour'`, - ); - const stalled = Number(rows[0]?.stalled ?? 0); - checks.push({ - name: 'queue_health', - status: stalled === 0 ? 'ok' : 'warn', - message: stalled === 0 - ? 'No stalled active jobs' - : `${stalled} active job(s) stalled > 1h — \`gbrain jobs cancel \` or \`gbrain jobs retry \` on the host`, - }); - } catch { - checks.push({ name: 'queue_health', status: 'ok', message: 'No queue activity' }); - } - } else { - checks.push({ name: 'queue_health', status: 'ok', message: 'PGLite — no queue to check' }); - } + checks.push(await computeQueueHealthCheck(engine)); // issue #1801 — wedged_queue (cross-surface parity with buildChecks). checks.push(await computeWedgedQueueCheck(engine)); @@ -1585,6 +1561,174 @@ export async function checkRerankerHealth(engine: BrainEngine): Promise { * Also surfaces (codex M-10): runs resolveBulkRetryOpts(process.env) at * startup so bad GBRAIN_BULK_* config fails at doctor time, not first-retry. */ +/** + * queue_health: Postgres Minion queue diagnostics. + * + * Includes the original stalled/depth/memory/prompt checks plus the #2557 + * no-worker signal: old `embed-backfill` jobs waiting on a queue with no live + * registered worker for that queue. That catches the default deployment shape + * where `sync` enqueues deferred embedding work but the operator never started + * `gbrain jobs work` or a supervisor. + */ +export async function computeQueueHealthCheck( + engine: BrainEngine, + opts: { + waitingDepthThreshold?: number; + oldWaitingHours?: number; + readWorkers?: () => Array<{ queue: string }>; + } = {}, +): Promise { + if (engine.kind === 'pglite') { + return { + name: 'queue_health', + status: 'ok', + message: 'Skipped (PGLite — no multi-process worker surface)', + }; + } + + try { + // issue #1801: column is `status`, not `state` (schema.sql:780). + const stalledRows: Array<{ id: number; name: string; started_at: string }> = + await engine.executeRaw( + `SELECT id, name, started_at::text AS started_at + FROM minion_jobs + WHERE status = 'active' + AND started_at IS NOT NULL + AND started_at < now() - interval '1 hour' + ORDER BY started_at ASC + LIMIT 5`, + ); + + const threshold = opts.waitingDepthThreshold + ?? _resolveEnvNumber('GBRAIN_QUEUE_WAITING_THRESHOLD', 10); + const depthRows: Array<{ name: string; queue: string; depth: number }> = + await engine.executeRaw( + `SELECT name, queue, count(*)::int AS depth + FROM minion_jobs + WHERE status = 'waiting' + GROUP BY name, queue + HAVING count(*) > $1 + ORDER BY depth DESC + LIMIT 5`, + [threshold], + ); + + const rssKillRows: Array<{ cnt: number }> = await engine.executeRaw( + `SELECT count(*)::int AS cnt + FROM minion_jobs + WHERE status IN ('dead', 'failed') + AND finished_at > now() - interval '24 hours' + AND error_text = 'aborted: watchdog'`, + ); + const rssKillCount = Number(rssKillRows[0]?.cnt ?? 0); + + const promptTooLongRows: Array<{ cnt: number }> = await engine.executeRaw( + `SELECT count(*)::int AS cnt + FROM minion_jobs + WHERE name = 'subagent' + AND status = 'dead' + AND finished_at > now() - interval '24 hours' + AND error_text LIKE 'prompt_too_long:%'`, + ); + const promptTooLongCount = Number(promptTooLongRows[0]?.cnt ?? 0); + + const oldWaitingHours = opts.oldWaitingHours + ?? _resolveEnvNumber('GBRAIN_QUEUE_NO_WORKER_WARN_HOURS', 1); + const oldWaitingRows: Array<{ + name: string; + queue: string; + depth: number; + oldest_age_seconds: number; + }> = await engine.executeRaw( + `SELECT name, + queue, + count(*)::int AS depth, + EXTRACT(EPOCH FROM (now() - min(created_at)))::int AS oldest_age_seconds + FROM minion_jobs + WHERE status = 'waiting' + AND name = 'embed-backfill' + GROUP BY name, queue + HAVING min(created_at) < now() - ($1::text::interval) + ORDER BY oldest_age_seconds DESC + LIMIT 5`, + [`${oldWaitingHours} hours`], + ); + + let liveWorkerQueues = new Set(); + if (oldWaitingRows.length > 0) { + const workers = opts.readWorkers + ? opts.readWorkers() + : (await import('../core/minions/worker-registry.ts')).readWorkers(); + liveWorkerQueues = new Set(workers.map((w) => w.queue)); + } + + const problems: string[] = []; + if (stalledRows.length > 0) { + const sample = stalledRows + .map(r => `#${r.id}(${r.name})`) + .join(', '); + problems.push( + `${stalledRows.length} stalled-forever job(s): ${sample}. ` + + `Fix: gbrain jobs get to inspect; gbrain jobs cancel to force-kill.` + ); + } + if (depthRows.length > 0) { + const sample = depthRows + .map(r => `${r.name}@${r.queue}=${r.depth}`) + .join(', '); + problems.push( + `waiting-queue depth exceeds ${threshold} for: ${sample}. ` + + `Fix: set maxWaiting on the submitter (or raise GBRAIN_QUEUE_WAITING_THRESHOLD).` + ); + } + for (const row of oldWaitingRows) { + if (liveWorkerQueues.has(row.queue)) continue; + const hours = Math.max(1, Math.round(Number(row.oldest_age_seconds ?? 0) / 3600)); + problems.push( + `${row.depth} ${row.name} job(s) have waited on queue '${row.queue}' for up to ${hours}h ` + + `and no live worker is registered for that queue. ` + + `Start one with \`gbrain jobs work --queue ${row.queue}\` or ` + + `\`gbrain jobs supervisor start --queue ${row.queue}\`.` + ); + } + if (rssKillCount > 0) { + problems.push( + `${rssKillCount} job(s) dead-lettered for RSS-watchdog memory-limit kills in last 24h. ` + + `Fix: raise the limit (e.g. \`gbrain jobs work --max-rss 4096\`) or opt out (\`--max-rss 0\`). ` + + `→ see worker_oom_loop for the cap + fix (the authoritative OOM-loop signal).` + ); + } + if (promptTooLongCount > 0) { + problems.push( + `${promptTooLongCount} subagent job(s) dead-lettered with prompt_too_long in last 24h. ` + + `Dream/synthesize transcripts exceeded the model's input context. ` + + `Fix: \`gbrain dream --phase synthesize --dry-run --json\` to identify fat transcripts; ` + + `set \`dream.synthesize.max_prompt_tokens\` to bound the per-chunk budget, or use a ` + + `larger-context model (Opus 4.7 = 1M tokens vs Sonnet 4.6 = 200K).` + ); + } + + if (problems.length === 0) { + return { + name: 'queue_health', + status: 'ok', + message: `No stalled-forever jobs; no queue over depth ${threshold}; no old embed-backfill jobs without a worker.`, + }; + } + return { + name: 'queue_health', + status: 'warn', + message: problems.join(' '), + }; + } catch (e) { + return { + name: 'queue_health', + status: 'warn', + message: `queue_health scan skipped: ${e instanceof Error ? e.message : String(e)}`, + }; + } +} + /** * issue #1801 — `wedged_queue` check. Surfaces the alive-but-wedged-worker * signature (a queue with claimable work waiting, zero live-lock active jobs, @@ -6915,159 +7059,12 @@ export async function buildChecks( } } - // 11b. Queue health (v0.19.1 queue-resilience wave). - // Postgres-only because PGLite has no multi-process worker surface. Two - // subchecks, both cheap (single SELECT each, status-index-covered): - // - // 1. stalled-forever: any active job whose started_at is > 1h old. The - // incident that motivated this release ran 90+ min before surfacing. - // Surface the ID so the operator can `gbrain jobs get ` to inspect - // or `gbrain jobs cancel ` to force-kill. - // - // 2. backpressure-missed: per-name waiting depth exceeds the threshold - // (default 10, override via GBRAIN_QUEUE_WAITING_THRESHOLD env). Signal - // that a submitter probably needs maxWaiting set. Bounded by per-name - // aggregation so a single name's pile shows up clearly instead of - // getting lost in the total. - // - // Not included in v0.19.1 (tracked as B7 follow-up): worker-heartbeat - // staleness. It needs a minion_workers table; the lock_until-on-active-jobs - // proxy can't distinguish "no worker" from "worker idle," and a check that - // cries wolf erodes trust in every other doctor check. progress.heartbeat('queue_health'); - if (engine.kind === 'pglite') { - checks.push({ - name: 'queue_health', - status: 'ok', - message: 'Skipped (PGLite — no multi-process worker surface)', - }); - } else { - const queueHealthHb = startHeartbeat(progress, 'scanning queue health…'); - try { - const sql = db.getConnection(); - // Subcheck 1: stalled-forever active jobs (>1h wall-clock). - const stalledRows: Array<{ id: number; name: string; started_at: string }> = await sql` - SELECT id, name, started_at::text AS started_at - FROM minion_jobs - WHERE status = 'active' - AND started_at IS NOT NULL - AND started_at < now() - interval '1 hour' - ORDER BY started_at ASC - LIMIT 5 - `; - // Subcheck 2: per-name waiting depth exceeds threshold. - const rawThreshold = process.env.GBRAIN_QUEUE_WAITING_THRESHOLD; - const parsedThreshold = rawThreshold ? parseInt(rawThreshold, 10) : 10; - const threshold = Number.isFinite(parsedThreshold) && parsedThreshold >= 1 - ? parsedThreshold - : 10; - const depthRows: Array<{ name: string; queue: string; depth: number }> = await sql` - SELECT name, queue, count(*)::int AS depth - FROM minion_jobs - WHERE status = 'waiting' - GROUP BY name, queue - HAVING count(*) > ${threshold} - ORDER BY depth DESC - LIMIT 5 - `; - // Subcheck 3 (v0.22.14): RSS-watchdog kills in the last 24h. Bare workers - // newly default to --max-rss 2048 (was 0); operators who run large embed - // or import jobs may see kills that didn't happen pre-v0.22.14. We surface - // a hint when this signature appears so the upgrade path is obvious. - // Signature: when the watchdog trips, gracefulShutdown('watchdog') aborts - // in-flight jobs with `new Error('watchdog')`. The worker's failJob path - // (worker.ts:660-664) writes `error_text = 'aborted: watchdog'` for any - // job in-flight at the moment of the kill. - // - // We deliberately DO NOT do a loose `ILIKE '%watchdog%'`: - // 1. Parent jobs that inherit `on_child_fail='fail_parent'` get - // `"child job N failed: aborted: watchdog"` — counting that - // double-counts (child + parent) for one watchdog event. - // 2. Any user error_text containing the word "watchdog" matches. - // Match the exact prefix `'aborted: watchdog'` to scope this purely to - // the worker's own kill signature. - const rssKillRows: Array<{ cnt: number }> = await sql` - SELECT count(*)::int AS cnt - FROM minion_jobs - WHERE status IN ('dead', 'failed') - AND finished_at > now() - interval '24 hours' - AND error_text = 'aborted: watchdog' - `; - const rssKillCount = rssKillRows[0]?.cnt ?? 0; - - // Subcheck 4 (v0.30.2): prompt_too_long terminal failures on subagent - // jobs in the last 24h. The dream/synthesize phase classifies Anthropic - // 400 "prompt is too long" responses as UnrecoverableError so they - // dead-letter on first attempt instead of clogging the queue with - // max_stalled retries. Surface count + fix hint when present. - const promptTooLongRows: Array<{ cnt: number }> = await sql` - SELECT count(*)::int AS cnt - FROM minion_jobs - WHERE name = 'subagent' - AND status = 'dead' - AND finished_at > now() - interval '24 hours' - AND error_text LIKE 'prompt_too_long:%' - `; - const promptTooLongCount = promptTooLongRows[0]?.cnt ?? 0; - - const problems: string[] = []; - if (stalledRows.length > 0) { - const sample = stalledRows - .map(r => `#${r.id}(${r.name})`) - .join(', '); - problems.push( - `${stalledRows.length} stalled-forever job(s): ${sample}. ` + - `Fix: gbrain jobs get to inspect; gbrain jobs cancel to force-kill.` - ); - } - if (depthRows.length > 0) { - const sample = depthRows - .map(r => `${r.name}@${r.queue}=${r.depth}`) - .join(', '); - problems.push( - `waiting-queue depth exceeds ${threshold} for: ${sample}. ` + - `Fix: set maxWaiting on the submitter (or raise GBRAIN_QUEUE_WAITING_THRESHOLD).` - ); - } - if (rssKillCount > 0) { - problems.push( - `${rssKillCount} job(s) dead-lettered for RSS-watchdog memory-limit kills in last 24h. ` + - `Fix: raise the limit (e.g. \`gbrain jobs work --max-rss 4096\`) or opt out (\`--max-rss 0\`). ` + - `→ see worker_oom_loop for the cap + fix (the authoritative OOM-loop signal).` - ); - } - if (promptTooLongCount > 0) { - problems.push( - `${promptTooLongCount} subagent job(s) dead-lettered with prompt_too_long in last 24h. ` + - `Dream/synthesize transcripts exceeded the model's input context. ` + - `Fix: \`gbrain dream --phase synthesize --dry-run --json\` to identify fat transcripts; ` + - `set \`dream.synthesize.max_prompt_tokens\` to bound the per-chunk budget, or use a ` + - `larger-context model (Opus 4.7 = 1M tokens vs Sonnet 4.6 = 200K).` - ); - } - - if (problems.length === 0) { - checks.push({ - name: 'queue_health', - status: 'ok', - message: `No stalled-forever jobs; no queue over depth ${threshold}.`, - }); - } else { - checks.push({ - name: 'queue_health', - status: 'warn', - message: problems.join(' '), - }); - } - } catch (e) { - checks.push({ - name: 'queue_health', - status: 'warn', - message: `queue_health scan skipped: ${e instanceof Error ? e.message : String(e)}`, - }); - } finally { - queueHealthHb(); - } + const queueHealthHb = startHeartbeat(progress, 'scanning queue health…'); + try { + checks.push(await computeQueueHealthCheck(engine)); + } finally { + queueHealthHb(); } // 11.4 subagent_capability (v0.38 — D7; was subagent_provider in v0.31.12). Surfaces a diff --git a/test/doctor-wedged-queue.test.ts b/test/doctor-wedged-queue.test.ts index c69932f1a..c22601a43 100644 --- a/test/doctor-wedged-queue.test.ts +++ b/test/doctor-wedged-queue.test.ts @@ -14,7 +14,7 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'bun:test' import { readFileSync } from 'fs'; import { join } from 'path'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; -import { computeWedgedQueueCheck } from '../src/commands/doctor.ts'; +import { computeQueueHealthCheck, computeWedgedQueueCheck } from '../src/commands/doctor.ts'; import type { BrainEngine } from '../src/core/engine.ts'; let base: PGLiteEngine; @@ -44,15 +44,43 @@ async function seed( queue: string, name: string, status: string, - extra: { lockUntilSql?: string; updatedAtSql?: string } = {}, + extra: { lockUntilSql?: string; updatedAtSql?: string; createdAtSql?: string } = {}, ): Promise { await base.executeRaw( - `INSERT INTO minion_jobs (name, queue, status, lock_until, updated_at) - VALUES ($1, $2, $3, ${extra.lockUntilSql ?? 'NULL'}, ${extra.updatedAtSql ?? 'now()'})`, + `INSERT INTO minion_jobs (name, queue, status, lock_until, updated_at, created_at) + VALUES ($1, $2, $3, ${extra.lockUntilSql ?? 'NULL'}, ${extra.updatedAtSql ?? 'now()'}, ${extra.createdAtSql ?? 'now()'})`, [name, queue, status], ); } +describe('issue #2557 — queue_health catches deferred embed with no worker', () => { + it('warns when old embed-backfill jobs have no live worker for their queue', async () => { + await seed('default', 'embed-backfill', 'waiting', { + createdAtSql: "now() - interval '3 hours'", + }); + const check = await computeQueueHealthCheck(pgLike, { + readWorkers: () => [], + oldWaitingHours: 1, + }); + expect(check.status).toBe('warn'); + expect(check.message).toContain('embed-backfill'); + expect(check.message).toContain('no live worker'); + expect(check.message).toContain('gbrain jobs work --queue default'); + }); + + it('does not warn for old embed-backfill jobs when a worker is live on that queue', async () => { + await seed('default', 'embed-backfill', 'waiting', { + createdAtSql: "now() - interval '3 hours'", + }); + const check = await computeQueueHealthCheck(pgLike, { + readWorkers: () => [{ queue: 'default' }], + oldWaitingHours: 1, + }); + expect(check.status).toBe('ok'); + expect(check.message).toContain('no old embed-backfill jobs without a worker'); + }); +}); + describe('issue #1801 fix #3 — computeWedgedQueueCheck', () => { it('flags a wedged queue (waiting, 0 active_healthy, stale completion) as fail', async () => { await seed('default', 'cycle', 'waiting');