mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 21:19:18 +00:00
* feat(jobs): niceness core, worker registry, shared supervisor-pid reader OS scheduling-priority primitives for issue #1815: - niceness.ts: parseNiceValue (whole-string), applyNiceness (re-reads effective in success AND failure paths), getEffectiveNiceness, formatNice. - worker-registry.ts: live workers self-register pid + requested/effective nice under gbrainPath('workers'); readWorkers prunes ESRCH (keeps EPERM) with a pid-reuse start-time guard. - supervisor-pid.ts: readSupervisorPid extracted from the copy-pasted PID-file + liveness block. * feat(jobs): --nice flag for jobs work/supervisor + doctor niceness check Wires the --nice <n> flag (and GBRAIN_NICE env) through the CLI (issue #1815): - jobs work: applies niceness + registers the worker; cleanup on finally and process.on('exit'). - jobs supervisor: applies in the foreground-start path only (after the --detach fork), passes the apply result into MinionSupervisor. - supervisor.ts: nice opts, extracted testable buildWorkerArgs (appends --nice), emits niceness on started/worker_spawned audit events. - jobs stats / supervisor status: surface effective worker + supervisor nice. - doctor: separate supervisor_niceness check (warns on requested != effective) so it can't clobber the supervisor crash-check precedence; registered in doctor-categories. * test(jobs): cover niceness, worker registry, supervisor-pid, build args Unit tests for issue #1815: parseNiceValue rejects 3.5/10abc that parseInt would accept; applyNiceness re-reads effective on EPERM; registry ESRCH/EPERM + pid-reuse guard + brain-isolated path; readSupervisorPid states; parseNiceFlag flag>env precedence; buildWorkerArgs --nice propagation. * chore: bump version and changelog (v0.42.23.0) --nice flag for jobs work/supervisor (issue #1815). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * docs: document --nice flag for jobs work/supervisor (v0.42.23.0) - minions-deployment.md: niceness tuning section (full concurrency, low priority). - KEY_FILES.md: entries for niceness.ts, worker-registry.ts, supervisor-pid.ts; supervisor.ts entry notes buildWorkerArgs + nice opts. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(e2e): add enrich_thin to dream cycle EXPECTED_PHASES The enrich_thin cycle phase (src/core/cycle.ts ALL_PHASES, between conversation_facts_backfill and skillopt) shipped without updating the e2e phase-order expectation, so dream-cycle-phase-order-pglite failed on master. Sync the expected list to the real ALL_PHASES order. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(e2e): align sync-lock-recovery with the shipped --break-lock --all contract v0.41.13.0 intentionally dropped the "--break-lock + --all is refused" guard so cron can self-heal every source in one call (sync.ts runBreakLock iterates sources under --all). The e2e test still asserted the old exit-1 refusal and failed on master. Assert the current contract: the combination is accepted and takes the iterate / no-active-sources path (exit 0, no refusal message). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(e2e): de-flake ingestion-roundtrip chokidar first-drop race The native fsevents watcher occasionally missed a freshly written file, timing out the 15s waitFor (~1/3 on master under load). Three fixes: - inject a polling chokidar watcher via the source's _watchFactory seam (usePolling, 20ms interval) so detection never depends on fsevents timing; - drop deterministic fixtures BEFORE start so the initial scan (ignoreInitial:false) emits them, keeping live-watch coverage only where it's robust; - poll for the dedup hit instead of a fixed 600ms sleep. 15/15 green under stress. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(e2e): make hermetic-PGLite serve tests actually hermetic connect-bearer and serve-stdio-roundtrip init a PGLite brain and spawn serve, but passed {...process.env} through — leaking an ambient DATABASE_URL / GBRAIN_DATABASE_URL into the subprocess, which then came up on Postgres and failed the `engine: pglite` assertion. Strip both DB vars from the spawned env so the tests are deterministic whether or not the shell/CI has a DB URL set. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(e2e): type the hermetic-PGLite env so tsc passes The DATABASE_URL/GBRAIN_DATABASE_URL strip used `delete` on a narrowly-typed env literal (tsc-only failure; bun test doesn't typecheck). Annotate connect-bearer's env as Record<string,string|undefined> and build serve-stdio's as a concrete Record<string,string> (StdioClientTransport.env rejects undefined). Runtime behavior unchanged (7/7 + 3/3 green). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test: quarantine worker-registry to *.serial (R1 env-mutation isolation) worker-registry.test.ts sets process.env.GBRAIN_HOME per-test so gbrainPath resolves to a temp dir, then lazy-imports the module — a process-global mutation the parallel isolation lint (rule R1) forbids. Rename to worker-registry.serial.test.ts: it runs in the serial pass (own bun process, max-concurrency=1) where env mutation is safe, and the lint skips *.serial files. No logic change (6/6 green); fixes the failing `verify` CI job. 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
f4959348c2
commit
f11d56cfca
@@ -2,6 +2,23 @@
|
||||
|
||||
All notable changes to GBrain will be documented in this file.
|
||||
|
||||
## [0.42.23.0] - 2026-06-03
|
||||
|
||||
**`gbrain jobs work` and `gbrain jobs supervisor` take a `--nice <n>` flag that lowers the background job tree's CPU scheduling priority without cutting concurrency.** When the Minions worker pool runs at full width (sync, embed, extract, subagent fans), it can drive a machine's load average high enough to starve your interactive shell. Dropping concurrency throws away throughput. Niceness is the right lever: keep full concurrency, run at low priority, and the work finishes just as fast when the box is idle while yielding politely when it's busy. In the real incident that drove this, reniceing the tree took load from ~7 to ~3 with no measurable throughput loss.
|
||||
|
||||
### What changed
|
||||
|
||||
- **`--nice <n>` on `gbrain jobs work` and `gbrain jobs supervisor`** (POSIX `-20`..`19`; positive = nicer/lower priority). Also reads `GBRAIN_NICE` (the flag wins over the env var).
|
||||
- **Propagates down the whole tree.** The supervisor renices itself and passes `--nice` to the worker it spawns; OS niceness inherits to the worker's own children (shell jobs, subagents) automatically.
|
||||
- **Effective niceness is observable.** `gbrain jobs stats` and `gbrain jobs supervisor status --json` report the live worker + supervisor niceness, and a new `supervisor_niceness` check in `gbrain doctor` surfaces it as structured data — warning when what you asked for isn't what's running (negative nice without privilege, or an OS `RLIMIT_NICE` clamp).
|
||||
|
||||
## To take advantage of v0.42.23.0
|
||||
|
||||
`gbrain upgrade`, then start your worker or supervisor with `--nice 10` (or set `GBRAIN_NICE=10`). Confirm with `gbrain jobs stats` or `gbrain doctor` — both report the effective value, and `ps -o ni` will agree. Positive values need no privilege; negative values (raising priority) need root. This is distinct from the concurrency / inflight cap and composes with it: `--nice` tunes *priority*, concurrency tunes *width*.
|
||||
|
||||
### For contributors
|
||||
|
||||
New `src/core/minions/niceness.ts` (`applyNiceness` re-reads the effective value in both the success and failure paths, so a denied renice records the real inherited value, not null), `worker-registry.ts` (live workers self-register under `gbrainPath('workers')`, brain-isolated, with `ESRCH`/`EPERM`-aware pruning and a pid-reuse start-time guard), and `supervisor-pid.ts` (shared PID-file reader, dedupes the status/doctor/stats copies). `buildWorkerArgs` was extracted from the supervisor for unit testing. Eng review + Codex outside-voice both cleared the plan; Codex caught the detached-supervisor renice-ordering bug and the `parseInt("3.5")` gap. Closes #1815.
|
||||
## [0.42.22.0] - 2026-06-03
|
||||
|
||||
**A background worker whose database connection quietly dies no longer sits there alive-but-doing-nothing for hours. Your brain keeps processing jobs instead of silently stalling overnight.**
|
||||
|
||||
@@ -211,9 +211,12 @@ per-release `**vX.Y.Z:**` narration — CI enforces this
|
||||
- `src/core/minions/` — Minions job queue: BullMQ-inspired, Postgres-native (queue, worker, backoff, types, protected-names, quiet-hours, stagger, handlers/shell).
|
||||
- `src/core/minions/queue.ts` — MinionQueue class (submit, claim, complete, fail, stall detection, parent-child, depth/child-cap, per-job timeouts, cascade-kill, attachments, idempotency keys, child_done inbox, removeOnComplete/Fail). `add()` takes a 4th `trusted` arg (separate from `opts` to prevent spread leakage); protected names in `PROTECTED_JOB_NAMES` require `{allowProtectedSubmit: true}` and the check runs trim-normalized (whitespace-bypass safe). `add()` plumbs `max_stalled` through with a `[1, 100]` clamp; omitted values let the schema DEFAULT (5) kick in. `handleWallClockTimeouts(lockDurationMs)` is Layer 3 kill shot for jobs where `FOR UPDATE SKIP LOCKED` stall detection and the timeout sweep both fail to evict (wedged worker holding a row lock via a pending transaction). The `maxWaiting` coalesce path uses `pg_advisory_xact_lock` keyed on `(name, queue)` to serialize concurrent submits for the same key, and filters on `queue` in addition to `name` so cross-queue same-name jobs don't suppress each other.
|
||||
- `src/core/minions/worker.ts` — MinionWorker class (handler registry, lock renewal, graceful shutdown, timeout safety net). Aborted jobs call `failJob` with reason (`timeout`/`cancel`/`lock-lost`/`shutdown`); `shutdownAbort` (instance field) fires on SIGTERM/SIGINT and propagates to `ctx.shutdownSignal` (shell handler listens; non-shell handlers don't). Per-job timeout fires `abort.abort(new Error('timeout'))` then a 30s grace-then-evict safety net force-evicts the job from `inFlight` and marks it dead if the handler ignores the abort signal (the moved-out generic abort listener now fires for ANY abort reason). The `launchJob` lock-renewal block is a thin sync wrapper around the pure `runLockRenewalTick` from `src/core/minions/lock-renewal-tick.ts` (NEVER `setInterval(async () => await renewLock(...))` — that shape was the unhandledRejection crash class during PgBouncer rotation). Gaps closed: (1) `cancelled` flag captured in the timer closure stops in-flight IIFEs writing misleading audit events after the job ended; (2) re-entrancy guard `tickInFlight` + per-call `Promise.race` timeout (`GBRAIN_LOCK_RENEWAL_CALL_TIMEOUT_MS`, default `lockDuration/3`); (3) time-based abort (`Date.now() - lastSuccessfulRenewalAt >= lockDuration - GBRAIN_LOCK_RENEWAL_SAFETY_MARGIN_MS`) so we release the lock BEFORE another worker can reclaim; (4) explicit `.catch()` on the stored `executeJob(...).finally(...)` promise closes the second unhandledRejection vector; (5) exported `INFRASTRUCTURE_ABORT_REASONS = new Set(['lock-renewal-failed', 'lock-lost'])` so executeJob's catch skips `failJob` for these (PgBouncer blips don't dead-letter healthy jobs; the stall detector reclaims). CI guard `scripts/check-worker-lock-renewal-shape.sh` (in `bun run verify`) asserts the bug pattern stays absent AND `launchJob` keeps calling `runLockRenewalTick`. Engine-ownership invariant: `start()` does NOT call `engine.disconnect()` on shutdown — the CLI handler in `src/commands/jobs.ts case 'work'` owns engine lifecycle via try/finally with loud error logging. RSS watchdog uses non-file-backed pages on Linux: exported `parseRssFromProcStatus(status)` (pure parser; field-presence regex so `RssAnon: 0 + RssShmem: 512` parses correctly) and `getAccurateRss(readStatus?)` (reads `/proc/self/status` for `RssAnon + RssShmem`, falls back to `process.memoryUsage().rss` on macOS / restricted containers / kernel <4.5); the default `getRss` in `WorkerOpts` is `getAccurateRss`. `checkMemoryLimit` tracks peak RSS, fires an 80%-of-cap soft-warn (once per crossing, carrying peak + in-flight job kinds), and on exceed sets `_rssWatchdogTriggered=true` (exposed via `get rssWatchdogTriggered()`) so `jobs work`'s finally can `process.exit(WORKER_EXIT_RSS_WATCHDOG)` after disconnect (drain self-identifying instead of an opaque code-0 exit). The poll loop wraps `claim` in try/catch: on a retryable conn error it reconnects ONCE and continues to the next tick rather than blind-retrying (a retry after `UPDATE...RETURNING` committed but the socket died would double-claim). `LockRenewalDeps` is wired with `reconnect` when the engine supports it. The self-health DB-liveness probe now runs EVEN under a supervisor (`GBRAIN_SUPERVISED=1`): the outer guard is `if (this.opts.healthCheckInterval > 0)` and only the STALL-detection block is wrapped in `if (!isSupervisedChild)` — so a supervised worker whose own pool dies self-exits `unhealthy(db_dead)` after `dbFailExitAfter` probes (the supervisor watches a different connection and can't see this worker's dead pool), while the supervisor's progress watchdog owns forward-progress (#1801). Pinned by `test/worker-lock-renewal.test.ts` (18), `test/audit/lock-renewal-audit.test.ts` (11), `test/scripts/check-worker-lock-renewal-shape.test.ts` (5), `test/worker-shutdown-disconnect.test.ts` (asserts `disconnectSpy).not.toHaveBeenCalled()`), `test/worker-rss.test.ts` (11), `test/worker-supervised-db-probe.test.ts` (3).
|
||||
- `src/core/minions/supervisor.ts` — MinionSupervisor process manager. Spawns `gbrain jobs work` as a child, restarts on crash with exponential backoff, periodic health check. `consecutiveHealthFailures` counter; on 3 consecutive failures emits `health_warn` with `reason: 'db_connection_degraded'` and calls `engine.reconnect()` to swap in a fresh pool, then resets. Worker exit classifier emits `likely_cause` on `worker_exited` events: `oom_or_external_kill` (SIGKILL), `graceful_shutdown` (SIGTERM), `runtime_error` (code 1), `clean_exit` (code 0), `unknown`. Consumes `detectTini()` + `buildSpawnInvocation()` from `src/core/minions/spawn-helpers.ts` to wrap the worker subtree in tini-as-PID-1 when tini is on `PATH` (handles native-addon zombie reaping the in-process SIGCHLD reaper can't reach); exposes `isTiniDetected` read-only accessor. The spawn-and-respawn loop is the shared `ChildWorkerSupervisor` core: MinionSupervisor composes it via `runSuperviseLoop()` → `new ChildWorkerSupervisor({...})` and maps `ChildSupervisorEvent` back through `emit()` SupervisorEvent (JSONL audit consumers see byte-compatible output). PID lock, signal handlers, health check, and `process.exit` on max-crashes stay in MinionSupervisor. `code=0` leaves `crashCount` untouched (so a worker alternating real crashes + watchdog drains still trips `max_crashes`); `cleanRestartBudget` (default 10 restarts per 60s) caps the macOS/non-Linux-fallback tight-loop via `health_warn { reason: 'clean_restart_budget_exceeded' }` + backoff. `shutdown()` drains via `childSupervisor.killChild('SIGTERM')` + `awaitChildExit(35_000)`. Progress watchdog (#1801): `healthCheck()` restarts an alive-but-wedged child via `childSupervisor.restartCurrentChild(35_000)` when a queue has claimable work, 0 live-lock active jobs, and stale completions across `wedgeRestartChecks` (default 3) consecutive checks past `wedgeRestartMinutes` (default 15, 0 disables) + a `startupGraceMs` window; bounded by `wedgeRestartLoopBudget` (default 3 / `wedgeRestartLoopWindowMs`) which switches to a one-shot `wedge_restart_loop` alert. The wedge query is the exported `queryWedgeSignals(engine, queue, handlerNames)` — name+queue-scoped, `active_healthy` = live-lock only (an expired-lock active row does NOT mask the wedge), due-delayed counted. Claimable names are derived at start via a throwaway `registerBuiltinHandlers` worker (its new `quiet` opt). Flags `--wedge-restart-minutes` / `--wedge-restart-checks` + env `GBRAIN_WEDGE_RESTART_MINUTES` / `GBRAIN_WEDGE_RESTART_CHECKS`. Pinned by `test/supervisor.test.ts` (16 cases), `test/supervisor-tini.test.ts`, and `test/supervisor-wedge.test.ts`.
|
||||
- `src/core/minions/supervisor.ts` — MinionSupervisor process manager. Spawns `gbrain jobs work` as a child, restarts on crash with exponential backoff, periodic health check. `consecutiveHealthFailures` counter; on 3 consecutive failures emits `health_warn` with `reason: 'db_connection_degraded'` and calls `engine.reconnect()` to swap in a fresh pool, then resets. Worker exit classifier emits `likely_cause` on `worker_exited` events: `oom_or_external_kill` (SIGKILL), `graceful_shutdown` (SIGTERM), `runtime_error` (code 1), `clean_exit` (code 0), `unknown`. Consumes `detectTini()` + `buildSpawnInvocation()` from `src/core/minions/spawn-helpers.ts` to wrap the worker subtree in tini-as-PID-1 when tini is on `PATH` (handles native-addon zombie reaping the in-process SIGCHLD reaper can't reach); exposes `isTiniDetected` read-only accessor. The spawn-and-respawn loop is the shared `ChildWorkerSupervisor` core: MinionSupervisor composes it via `runSuperviseLoop()` → `new ChildWorkerSupervisor({...})` and maps `ChildSupervisorEvent` back through `emit()` SupervisorEvent (JSONL audit consumers see byte-compatible output). PID lock, signal handlers, health check, and `process.exit` on max-crashes stay in MinionSupervisor. `code=0` leaves `crashCount` untouched (so a worker alternating real crashes + watchdog drains still trips `max_crashes`); `cleanRestartBudget` (default 10 restarts per 60s) caps the macOS/non-Linux-fallback tight-loop via `health_warn { reason: 'clean_restart_budget_exceeded' }` + backoff. `shutdown()` drains via `childSupervisor.killChild('SIGTERM')` + `awaitChildExit(35_000)`. Progress watchdog (#1801): `healthCheck()` restarts an alive-but-wedged child via `childSupervisor.restartCurrentChild(35_000)` when a queue has claimable work, 0 live-lock active jobs, and stale completions across `wedgeRestartChecks` (default 3) consecutive checks past `wedgeRestartMinutes` (default 15, 0 disables) + a `startupGraceMs` window; bounded by `wedgeRestartLoopBudget` (default 3 / `wedgeRestartLoopWindowMs`) which switches to a one-shot `wedge_restart_loop` alert. The wedge query is the exported `queryWedgeSignals(engine, queue, handlerNames)` — name+queue-scoped, `active_healthy` = live-lock only (an expired-lock active row does NOT mask the wedge), due-delayed counted. Claimable names are derived at start via a throwaway `registerBuiltinHandlers` worker (its new `quiet` opt). Flags `--wedge-restart-minutes` / `--wedge-restart-checks` + env `GBRAIN_WEDGE_RESTART_MINUTES` / `GBRAIN_WEDGE_RESTART_CHECKS`. The worker argv is built by the exported pure `buildWorkerArgs(opts)` (appends `--nice N` when `opts.nice_requested` is set, alongside `--concurrency`/`--queue`/`--max-rss`); the niceness apply RESULT (`nice_requested`/`nice_effective`/`nice_error`, computed by the CLI in jobs.ts — the supervisor doesn't call setPriority) rides on the `started`/`worker_spawned` audit emissions (#1815). Pinned by `test/supervisor.test.ts` (16 cases), `test/supervisor-tini.test.ts`, `test/supervisor-wedge.test.ts`, and `test/supervisor-build-worker-args.test.ts`.
|
||||
- `src/core/minions/child-worker-supervisor.ts` — shared spawn-and-respawn core reused by both `MinionSupervisor` (standalone `gbrain jobs supervisor` daemon) and `src/commands/autopilot.ts` (autopilot daemon), so the two consumers can't drift into parallel-loop bugs. Pure class: NO PID file, NO signal handlers, NO `process.exit`, NO health check. Lifecycle events fire via injected `onEvent: (ChildSupervisorEvent) => void`. Exit classifier: `code === 0` leaves `crashCount` UNCHANGED (preserves flap detection across mixed exit sequences); `code != 0` follows `runDuration > stableRunResetMs ? 1 : ++crashCount`. Clean-restart budget: sliding window of code=0 exits; when count exceeds `cleanRestartBudget` (default 10) inside `cleanRestartWindowMs` (default 60s), emits `health_warn { reason: 'clean_restart_budget_exceeded' }` and applies `cleanRestartBudgetBackoffMs` (default 1s). The exit classifier special-cases `WORKER_EXIT_RSS_WATCHDOG` — `likely_cause='rss_watchdog'`, and that exit does NOT bump `crashCount` (routes to its own breaker); a dedicated `_watchdogExitTimestamps` sliding window trips a loud `rss_watchdog_loop` health_warn naming the cap when N watchdog exits land inside the window, INDEPENDENT of the stable-run reset (which would otherwise hide a >5-min-run watchdog drain loop). New opts `watchdogLoopBudget` (3), `watchdogLoopWindowMs` (600000), `watchdogBackoffMs` (30000); `ChildSupervisorEvent` extended. Public read-only accessors `childAlive`, `inBackoff`, `crashCount`; `killChild(signal)` gates on liveness (`exitCode === null && signalCode === null`), NOT `.killed` — `.killed` flips true once a signal is *sent*, so the old `!this._child.killed` guard made a follow-up SIGKILL after an ignored SIGTERM a silent no-op (#1801; the bug also lived in the `shutdown()` drain). `restartCurrentChild(graceMs)` (#1801 wedge self-heal) captures the CURRENT child ref, SIGTERM→grace→SIGKILLs THAT ref (never the respawn — closes the timer-kills-fresh-worker race), and flags `_intentionalRestart` so the exit is `likelyCause='wedge_restart'`, leaves `crashCount` UNTOUCHED (never trips `max_crashes`; like `rss_watchdog`), and respawns immediately (`backoff ms:0 reason='wedge_restart'`). `awaitChildExit(timeoutMs)` short-circuits when `child.exitCode !== null || child.signalCode !== null` so fast-SIGTERM responders don't cause a 35s shutdown hang. Test hooks `_backoffFloorMs`, `_now`. `supervisor-audit.ts` adds `rss_watchdog` as a non-clean cause + its own `CrashSummary.by_cause` bucket, and `wedge_restart` to `CLEAN_EXIT_CAUSES` (a self-heal, not a crash; denylist preserved so future causes route to `legacy`). Pinned by `test/child-worker-supervisor.test.ts` (12 cases).
|
||||
- `src/core/minions/spawn-helpers.ts` — pure `detectTini()` + `buildSpawnInvocation()` consumed by both `supervisor.ts` and `autopilot.ts` (resolves the DRY violation between the two spawn sites and makes tini wrapping testable without `mock.module()`, rule R2 of `scripts/check-test-isolation.sh`). `detectTini()` calls `execFileSync('which', ['tini'])` with explicit `env: process.env` so Bun sees runtime PATH mutations. `buildSpawnInvocation(tiniPath, cmd, args)` returns `{cmd, args}` with tini prepended when present, or the bare invocation otherwise. Pinned by `test/spawn-helpers.test.ts` (5) and `test/supervisor-tini.test.ts` (4).
|
||||
- `src/core/minions/niceness.ts` — OS scheduling-priority (niceness) primitives for the `--nice` flag. `parseNiceValue(raw)` whole-string parses + range-validates to POSIX `[-20, 19]` (rejects `"3.5"`/`"10abc"` that `parseInt` would silently truncate). `applyNiceness(nice, setPriority?, getPriority?)` calls `os.setPriority(0, n)` and ALWAYS re-reads `os.getPriority(0)` afterwards — in both the success and the catch paths — so a denied renice (EPERM) or an `RLIMIT_NICE` clamp records the real effective value (e.g. `0`), not null; returns `{applied, requested, effective, error?}`. `getEffectiveNiceness(pid, getPriority?)` reads an arbitrary pid's niceness (null on dead/unreadable). `formatNice(n)` → `+10`/`0`/`-5`. Applied only at the CLI layer (jobs.ts) so worker.ts/supervisor.ts stay embeddable. Pinned by `test/niceness.test.ts`.
|
||||
- `src/core/minions/worker-registry.ts` — live worker registry backing niceness observability. Each running `gbrain jobs work` self-registers `worker-<pid>.json` under `gbrainPath('workers')` (brain-isolated via `GBRAIN_HOME`; entries tagged with `currentBrainId()` so multiple DBs under one home don't cross-report). `registerWorker(info)` is best-effort (never blocks the worker) and returns a cleanup fn the caller wires to BOTH the shutdown `finally` AND `process.on('exit')` (the unhealthy `process.exit(1)` bypasses the awaited finally). `readWorkers(getNice?)` enumerates the dir, drops confirmed-dead pids (`classifyLiveness`: ESRCH = dead/prune, EPERM = alive/keep), applies a pid-reuse start-time guard (`ps -o lstart`, rejects a pid whose process started >5s after the entry was written), and re-measures each live worker's niceness now. Reports the worker's REAL pid, sidestepping the tini-wrapper-pid problem. Pinned by `test/worker-registry.test.ts`.
|
||||
- `src/core/minions/supervisor-pid.ts` — `readSupervisorPid(pidFile) → {pid, running}`: the shared `existsSync → readFileSync → parseInt → process.kill(pid,0)` PID-file + liveness reader extracted from the three copies in `jobs.ts` (supervisor status), `jobs.ts` (stats), and `doctor.ts`. EPERM from the liveness probe counts as running. Pinned by `test/supervisor-pid.test.ts`.
|
||||
- `src/core/minions/types.ts` — `MinionJobInput` + `MinionJobStatus` + handler context types. `MinionJobInput.max_stalled` is optional; omitted values let the schema DEFAULT (5) kick in, provided values are clamped to `[1, 100]`.
|
||||
- `src/core/minions/protected-names.ts` — side-effect-free constant module exporting `PROTECTED_JOB_NAMES` + `isProtectedJobName()`. Kept pure so queue core can import without loading handler modules.
|
||||
- `src/core/minions/handlers/shell.ts` — `shell` job handler. Spawns `/bin/sh -c cmd` (absolute path, PATH-override-safe) or `argv[0] argv[1..]` (no shell). Env allowlist `PATH, HOME, USER, LANG, TZ, NODE_ENV` + caller `env:` overrides + `inherit:`-resolved keys. UTF-8-safe stdout/stderr tail via `string_decoder.StringDecoder`. Abort (either `ctx.signal` or `ctx.shutdownSignal`) fires SIGTERM → 5s grace → SIGKILL on child. Requires `GBRAIN_ALLOW_SHELL_JOBS=1` on worker (gated by `registerBuiltinHandlers`). `ShellJobParams.inherit?: string[]` is a free-form list of snake_case config-key names; the worker resolves each via `loadConfig()` and injects the value under the derived env key (`database_url` → `GBRAIN_DATABASE_URL`; else uppercased). Names persist in `minion_jobs.data` (and the shell-audit JSONL); values never do. The canonical validator `validateShellJobParams` (sibling `shell-validate.ts`) runs PRE-ENQUEUE in both submit surfaces — `gbrain jobs submit shell` (jobs.ts:271) AND the `submit_job` op for `name='shell'` (operations.ts:2085); the handler-entry re-validation here is defense-in-depth (closes the bug class where validation ran AFTER `queue.add()` persisted the row). The validator does NOT police which config keys the agent inherits — same-uid trust model treats the agent as a peer of the worker.
|
||||
|
||||
@@ -54,6 +54,33 @@ gbrain jobs supervisor stop
|
||||
An agent seeing exit=2 can safely treat it as "one is already running";
|
||||
exit=1 should page a human.
|
||||
|
||||
### Lowering scheduling priority (`--nice`)
|
||||
|
||||
When the worker pool runs at full concurrency on a machine you also use
|
||||
interactively, it can drive the load average high enough to starve your
|
||||
shell. Cutting `--concurrency` throws away throughput. Reach for `--nice`
|
||||
instead — it lowers the job tree's CPU scheduling priority without touching
|
||||
width, so the work runs full-speed when the box is idle and yields when it
|
||||
isn't:
|
||||
|
||||
```bash
|
||||
# Full concurrency, low priority. Propagates to the spawned worker and its
|
||||
# children (shell jobs, subagents) via OS niceness inheritance.
|
||||
gbrain jobs supervisor --concurrency 4 --nice 10
|
||||
|
||||
# Equivalent for a bare worker, or set it durably in the environment.
|
||||
GBRAIN_NICE=10 gbrain jobs work --concurrency 4
|
||||
```
|
||||
|
||||
`--nice` takes a POSIX value from `-20` (highest priority) to `19`
|
||||
(nicest/lowest); positive values need no privilege, negative values need
|
||||
root. `GBRAIN_NICE` is the env equivalent (the flag wins). Confirm the
|
||||
effective value with `gbrain jobs stats`, `gbrain jobs supervisor status
|
||||
--json`, or the `supervisor_niceness` check in `gbrain doctor` — the doctor
|
||||
check warns if what you asked for isn't what's actually running (e.g. a
|
||||
negative value denied without privilege, or an OS `RLIMIT_NICE` clamp). This
|
||||
is distinct from the concurrency / inflight cap and composes with it.
|
||||
|
||||
### Which supervisor when?
|
||||
|
||||
The supervisor solves in-process crash recovery. Platform-level
|
||||
|
||||
+1
-1
@@ -143,5 +143,5 @@
|
||||
"bun": ">=1.3.10"
|
||||
},
|
||||
"license": "MIT",
|
||||
"version": "0.42.22.0"
|
||||
"version": "0.42.23.0"
|
||||
}
|
||||
|
||||
+62
-12
@@ -4136,19 +4136,11 @@ export async function buildChecks(
|
||||
try {
|
||||
const { DEFAULT_PID_FILE } = await import('../core/minions/supervisor.ts');
|
||||
const { readSupervisorEvents, summarizeCrashes } = await import('../core/minions/handlers/supervisor-audit.ts');
|
||||
const { readSupervisorPid } = await import('../core/minions/supervisor-pid.ts');
|
||||
|
||||
let supervisorPid: number | null = null;
|
||||
let running = false;
|
||||
if (existsSync(DEFAULT_PID_FILE)) {
|
||||
try {
|
||||
const line = readFileSync(DEFAULT_PID_FILE, 'utf8').trim().split('\n')[0];
|
||||
const parsed = parseInt(line, 10);
|
||||
if (!isNaN(parsed) && parsed > 0) {
|
||||
supervisorPid = parsed;
|
||||
try { process.kill(parsed, 0); running = true; } catch { running = false; }
|
||||
}
|
||||
} catch { /* unreadable */ }
|
||||
}
|
||||
const pidStatus = readSupervisorPid(DEFAULT_PID_FILE);
|
||||
const supervisorPid = pidStatus.pid;
|
||||
const running = pidStatus.running;
|
||||
|
||||
const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
|
||||
const lastStart = events.filter(e => e.event === 'started').pop()?.ts ?? null;
|
||||
@@ -4204,6 +4196,64 @@ export async function buildChecks(
|
||||
// Audit read / import failure is best-effort; skip silently.
|
||||
}
|
||||
|
||||
// 3b-sexies. Supervisor/worker scheduling priority (niceness, issue #1815).
|
||||
// SEPARATE check from `supervisor` above so a niceness divergence warn can
|
||||
// never clobber the supervisor check's max_crashes_exceeded fail/warn
|
||||
// precedence (Codex #11). Only surfaces when --nice was actually used (a live
|
||||
// worker exists or the supervisor recorded a niceness), so installs that never
|
||||
// touched --nice get no noise.
|
||||
try {
|
||||
const { DEFAULT_PID_FILE } = await import('../core/minions/supervisor.ts');
|
||||
const { readSupervisorPid } = await import('../core/minions/supervisor-pid.ts');
|
||||
const { readWorkers } = await import('../core/minions/worker-registry.ts');
|
||||
const { getEffectiveNiceness, formatNice } = await import('../core/minions/niceness.ts');
|
||||
|
||||
const sup = readSupervisorPid(DEFAULT_PID_FILE);
|
||||
const supervisorNice = sup.running && sup.pid !== null ? getEffectiveNiceness(sup.pid) : null;
|
||||
const workers = readWorkers().map(w => ({
|
||||
pid: w.pid,
|
||||
queue: w.queue,
|
||||
brain_id: w.brain_id,
|
||||
nice_requested: w.nice_requested,
|
||||
nice_effective: w.nice_now,
|
||||
}));
|
||||
|
||||
if (workers.length > 0 || supervisorNice !== null) {
|
||||
// Divergence: a worker (or the supervisor) asked for a niceness it didn't
|
||||
// get — usually negative nice without privilege, or an RLIMIT_NICE clamp.
|
||||
const diverged = workers.filter(
|
||||
w => w.nice_requested !== null && w.nice_effective !== null && w.nice_requested !== w.nice_effective,
|
||||
);
|
||||
|
||||
const workerSummary = workers
|
||||
.map(w => `pid ${w.pid}=${w.nice_effective !== null ? formatNice(w.nice_effective) : '?'}`)
|
||||
.join(', ');
|
||||
const supPart = supervisorNice !== null ? `supervisor=${formatNice(supervisorNice)}` : '';
|
||||
const okMsg = [supPart, workerSummary && `workers: ${workerSummary}`].filter(Boolean).join('; ');
|
||||
|
||||
if (diverged.length > 0) {
|
||||
const detail = diverged
|
||||
.map(w => `pid ${w.pid} requested ${formatNice(w.nice_requested!)} but running at ${formatNice(w.nice_effective!)}`)
|
||||
.join('; ');
|
||||
checks.push({
|
||||
name: 'supervisor_niceness',
|
||||
status: 'warn',
|
||||
message: `Niceness not applied as requested (${detail}). Negative nice needs privilege; the OS may also clamp to RLIMIT_NICE. Workers run at their inherited priority.`,
|
||||
details: { supervisor_nice: supervisorNice, workers },
|
||||
});
|
||||
} else {
|
||||
checks.push({
|
||||
name: 'supervisor_niceness',
|
||||
status: 'ok',
|
||||
message: okMsg || 'No niceness override active',
|
||||
details: { supervisor_nice: supervisorNice, workers },
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Registry / import failure is best-effort; skip silently.
|
||||
}
|
||||
|
||||
// 3b-quater. Worker OOM-loop (issue #1685 GAP A) — the single authoritative
|
||||
// "is the worker OOM-looping" line, unioning supervised (supervisor audit)
|
||||
// and bare-worker (minion_jobs watchdog-abort) kills. Returns null when the
|
||||
|
||||
+130
-17
@@ -10,6 +10,7 @@ import { WORKER_EXIT_RSS_WATCHDOG } from '../core/minions/worker-exit-codes.ts';
|
||||
import type { MinionJob, MinionJobStatus } from '../core/minions/types.ts';
|
||||
import { loadConfig, isThinClient } from '../core/config.ts';
|
||||
import { callRemoteTool, unpackToolResult } from '../core/mcp-client.ts';
|
||||
import { parseNiceValue, applyNiceness, getEffectiveNiceness, formatNice } from '../core/minions/niceness.ts';
|
||||
|
||||
function parseFlag(args: string[], flag: string): string | undefined {
|
||||
const idx = args.indexOf(flag);
|
||||
@@ -61,6 +62,22 @@ export function parseMaxRssFlag(args: string[]): number | undefined {
|
||||
return parsed;
|
||||
}
|
||||
|
||||
/** Parse `--nice N` (then `GBRAIN_NICE` env). Returns:
|
||||
* - undefined if absent (no priority change — inherit)
|
||||
* - the validated integer in [-20, 19] otherwise
|
||||
* Errors and exits the process on non-integer / out-of-range input (mirrors
|
||||
* parseMaxRssFlag's fail-fast). Flag wins over env. (issue #1815) */
|
||||
export function parseNiceFlag(args: string[], env: NodeJS.ProcessEnv = process.env): number | undefined {
|
||||
const raw = parseFlag(args, '--nice') ?? env.GBRAIN_NICE;
|
||||
if (raw === undefined || raw === '') return undefined;
|
||||
try {
|
||||
return parseNiceValue(raw);
|
||||
} catch (e) {
|
||||
console.error(`Error: ${e instanceof Error ? e.message : String(e)}`);
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
@@ -138,12 +155,19 @@ USAGE
|
||||
gbrain jobs stats
|
||||
gbrain jobs smoke
|
||||
gbrain jobs work [--queue Q] [--concurrency N] [--max-rss MB]
|
||||
[--health-interval MS]
|
||||
[--health-interval MS] [--nice N]
|
||||
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]
|
||||
[--max-rss MB] [--nice N]
|
||||
|
||||
--nice N OS scheduling priority, -20 (highest) to 19 (nicest). Lowers CPU
|
||||
priority without cutting concurrency — full throughput when the
|
||||
box is idle, yields to foreground work when it's busy. Propagates
|
||||
to spawned workers and their children. Env: GBRAIN_NICE (flag
|
||||
wins). Effective value shows in 'jobs stats' and 'gbrain doctor'.
|
||||
Negative values need root.
|
||||
gbrain jobs supervisor status [--json] [--pid-file PATH]
|
||||
gbrain jobs supervisor stop [--json] [--pid-file PATH]
|
||||
|
||||
@@ -548,6 +572,29 @@ HANDLER TYPES (built in)
|
||||
}
|
||||
console.log(`\n Queue health: ${stats.queue_health.waiting} waiting, ${stats.queue_health.active} active, ${stats.queue_health.stalled} stalled`);
|
||||
|
||||
// Scheduling priority (niceness, issue #1815). Best-effort: measures live
|
||||
// workers from the registry + the supervisor (if running) — silently skips
|
||||
// when nothing is reniced/running, so default stats output stays clean.
|
||||
try {
|
||||
const { readWorkers } = await import('../core/minions/worker-registry.ts');
|
||||
const { readSupervisorPid } = await import('../core/minions/supervisor-pid.ts');
|
||||
const { DEFAULT_PID_FILE } = await import('../core/minions/supervisor.ts');
|
||||
const liveWorkers = readWorkers();
|
||||
const sup = readSupervisorPid(DEFAULT_PID_FILE);
|
||||
const supNice = sup.running && sup.pid !== null ? getEffectiveNiceness(sup.pid) : null;
|
||||
if (liveWorkers.length > 0 || supNice !== null) {
|
||||
console.log(`\n Scheduling priority (nice):`);
|
||||
if (supNice !== null) console.log(` supervisor (pid ${sup.pid}): ${formatNice(supNice)}`);
|
||||
for (const w of liveWorkers) {
|
||||
const diverged = w.nice_requested !== null && w.nice_now !== null && w.nice_requested !== w.nice_now
|
||||
? ` ⚠ requested ${formatNice(w.nice_requested)}, not applied` : '';
|
||||
console.log(` worker (pid ${w.pid}, queue ${w.queue}): ${w.nice_now !== null ? formatNice(w.nice_now) : '?'}${diverged}`);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Registry/import failure is best-effort; skip silently.
|
||||
}
|
||||
|
||||
// issue #1801 — wedged-queue signature (queue-scoped): a worker is alive
|
||||
// but claiming nothing while work waits. `active_healthy` (live-lock only)
|
||||
// means an expired-lock active row doesn't mask it. Loud line so the
|
||||
@@ -838,6 +885,22 @@ HANDLER TYPES (built in)
|
||||
healthCheckInterval = parsed;
|
||||
}
|
||||
|
||||
// --nice N (issue #1815): renice this worker process so background work
|
||||
// yields CPU to foreground tasks without sacrificing concurrency. Applied
|
||||
// at the CLI layer (worker.ts stays embeddable). Niceness inherits to the
|
||||
// worker's spawned children (shell jobs / subagents) automatically.
|
||||
const niceVal = parseNiceFlag(args);
|
||||
let niceResult: ReturnType<typeof applyNiceness> | undefined;
|
||||
if (niceVal !== undefined) {
|
||||
niceResult = applyNiceness(niceVal);
|
||||
if (!niceResult.applied) {
|
||||
console.error(
|
||||
`[gbrain jobs] could not set niceness to ${niceVal}: ${niceResult.error ?? 'unknown'}. ` +
|
||||
`Negative nice needs privilege; running at niceness ${niceResult.effective ?? 'unchanged'}.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
try { await queue.ensureSchema(); }
|
||||
catch (e) { console.error(e instanceof Error ? e.message : String(e)); process.exit(1); }
|
||||
|
||||
@@ -882,11 +945,28 @@ HANDLER TYPES (built in)
|
||||
? `, db-probe: ${Math.round(healthCheckInterval / 1000)}s`
|
||||
: `, health-check: ${Math.round(healthCheckInterval / 1000)}s`)
|
||||
: '';
|
||||
console.log(`Minion worker started (queue: ${queueName}, concurrency: ${concurrency}${watchdogNote}${healthNote})`);
|
||||
const niceNote = niceResult ? `, nice: ${formatNice(niceResult.effective ?? niceVal!)}` : '';
|
||||
console.log(`Minion worker started (queue: ${queueName}, concurrency: ${concurrency}${watchdogNote}${healthNote}${niceNote})`);
|
||||
console.log(`Registered handlers: ${worker.registeredNames.join(', ')}`);
|
||||
|
||||
// Register in the live worker registry (issue #1815) so jobs stats / doctor
|
||||
// can report this worker's effective niceness. Cleanup runs on BOTH the
|
||||
// finally below AND process.on('exit') — the unhealthy handler's
|
||||
// process.exit(1) bypasses the awaited finally (Codex #10).
|
||||
const { registerWorker } = await import('../core/minions/worker-registry.ts');
|
||||
const unregisterWorker = registerWorker({
|
||||
pid: process.pid,
|
||||
queue: queueName,
|
||||
nice_requested: niceVal ?? null,
|
||||
nice_effective: niceResult ? niceResult.effective : null,
|
||||
started_at: Date.now(),
|
||||
});
|
||||
process.on('exit', () => unregisterWorker());
|
||||
|
||||
try {
|
||||
await worker.start();
|
||||
} finally {
|
||||
unregisterWorker();
|
||||
// Release the DB connection pool immediately on shutdown so
|
||||
// PgBouncer slots are freed rather than waiting for TCP keepalive
|
||||
// (~minutes). Disconnect failure is best-effort but logged loudly:
|
||||
@@ -933,21 +1013,13 @@ HANDLER TYPES (built in)
|
||||
|
||||
// ----- status subcommand -----
|
||||
if (isStatusCmd) {
|
||||
const { existsSync, readFileSync } = await import('fs');
|
||||
const { readSupervisorEvents, summarizeCrashes } = await import('../core/minions/handlers/supervisor-audit.ts');
|
||||
const { readSupervisorPid } = await import('../core/minions/supervisor-pid.ts');
|
||||
const { readWorkers } = await import('../core/minions/worker-registry.ts');
|
||||
|
||||
let supervisorPid: number | null = null;
|
||||
let running = false;
|
||||
if (existsSync(pidFile)) {
|
||||
try {
|
||||
const line = readFileSync(pidFile, 'utf8').trim().split('\n')[0];
|
||||
const parsed = parseInt(line, 10);
|
||||
if (!isNaN(parsed) && parsed > 0) {
|
||||
supervisorPid = parsed;
|
||||
try { process.kill(parsed, 0); running = true; } catch { running = false; }
|
||||
}
|
||||
} catch { /* unreadable PID file */ }
|
||||
}
|
||||
const pidStatus = readSupervisorPid(pidFile);
|
||||
const supervisorPid = pidStatus.pid;
|
||||
const running = pidStatus.running;
|
||||
|
||||
const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
|
||||
const lastStart = events.filter(e => e.event === 'started').pop()?.ts ?? null;
|
||||
@@ -958,6 +1030,17 @@ HANDLER TYPES (built in)
|
||||
const summary = summarizeCrashes(events);
|
||||
const maxCrashesEvent = events.filter(e => e.event === 'max_crashes_exceeded').pop() ?? null;
|
||||
|
||||
// Niceness (issue #1815): measure live workers + the supervisor itself.
|
||||
const workers = readWorkers().map(w => ({
|
||||
pid: w.pid,
|
||||
queue: w.queue,
|
||||
nice_requested: w.nice_requested,
|
||||
nice: w.nice_now,
|
||||
}));
|
||||
const supervisorNice = running && supervisorPid !== null
|
||||
? getEffectiveNiceness(supervisorPid)
|
||||
: null;
|
||||
|
||||
const status = {
|
||||
running,
|
||||
supervisor_pid: supervisorPid,
|
||||
@@ -967,6 +1050,8 @@ HANDLER TYPES (built in)
|
||||
clean_exits_24h: summary.clean_exits,
|
||||
crashes_by_cause: summary.by_cause,
|
||||
max_crashes_exceeded: !!maxCrashesEvent,
|
||||
nice: supervisorNice,
|
||||
workers,
|
||||
};
|
||||
|
||||
if (jsonMode) {
|
||||
@@ -978,6 +1063,12 @@ HANDLER TYPES (built in)
|
||||
if (lastStart) console.log(` Last start: ${lastStart}`);
|
||||
console.log(` Crashes (24h): ${summary.total} (runtime=${summary.by_cause.runtime_error} oom=${summary.by_cause.oom_or_external_kill} unknown=${summary.by_cause.unknown} legacy=${summary.by_cause.legacy})`);
|
||||
console.log(` Clean exits (24h): ${summary.clean_exits}`);
|
||||
if (supervisorNice !== null) console.log(` Nice (supervisor): ${formatNice(supervisorNice)}`);
|
||||
for (const w of workers) {
|
||||
const req = w.nice_requested !== null && w.nice !== null && w.nice_requested !== w.nice
|
||||
? ` (requested ${formatNice(w.nice_requested)})` : '';
|
||||
console.log(` Worker pid ${w.pid} [${w.queue}]: nice ${w.nice !== null ? formatNice(w.nice) : '?'}${req}`);
|
||||
}
|
||||
if (maxCrashesEvent) console.log(` ⚠ Max crashes exceeded at ${maxCrashesEvent.ts}`);
|
||||
}
|
||||
process.exit(running ? 0 : 1);
|
||||
@@ -1083,6 +1174,12 @@ HANDLER TYPES (built in)
|
||||
await import('../core/minions/rss-default.ts');
|
||||
const maxRssMb = parseMaxRssFlag(args) ?? resolveSupMaxRss();
|
||||
|
||||
// --nice N (issue #1815): validated here (fail-fast on bad input even for
|
||||
// --detach), but APPLIED only in the foreground-start path below — applying
|
||||
// before the --detach branch would renice the throwaway parent that forks
|
||||
// and exits, not the long-lived re-exec'd child (Codex #1).
|
||||
const supNice = parseNiceFlag(args);
|
||||
|
||||
const cliPath = parseFlag(args, '--cli-path') ?? resolveGbrainCliPath();
|
||||
|
||||
// --detach: fork a background supervisor, print PID payload, exit 0.
|
||||
@@ -1108,8 +1205,21 @@ HANDLER TYPES (built in)
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
// Foreground start.
|
||||
// Foreground start. Renice THIS process (the long-lived supervisor) now,
|
||||
// after the --detach fork-and-exit branch (Codex #1). The worker inherits
|
||||
// it via the spawn env; the supervisor also passes `--nice` down so the
|
||||
// worker re-applies it (see buildWorkerArgs).
|
||||
const supervisorPid = process.pid;
|
||||
let supNiceResult: ReturnType<typeof applyNiceness> | undefined;
|
||||
if (supNice !== undefined) {
|
||||
supNiceResult = applyNiceness(supNice);
|
||||
if (!supNiceResult.applied) {
|
||||
console.error(
|
||||
`[gbrain jobs] could not set supervisor niceness to ${supNice}: ${supNiceResult.error ?? 'unknown'}. ` +
|
||||
`Negative nice needs privilege; running at niceness ${supNiceResult.effective ?? 'unchanged'}.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
const supervisor = new MinionSupervisor(engine, {
|
||||
concurrency,
|
||||
queue: queueName,
|
||||
@@ -1120,6 +1230,9 @@ HANDLER TYPES (built in)
|
||||
allowShellJobs,
|
||||
json: jsonMode,
|
||||
maxRssMb,
|
||||
...(supNice !== undefined ? { nice_requested: supNice } : {}),
|
||||
...(supNiceResult?.effective != null ? { nice_effective: supNiceResult.effective } : {}),
|
||||
...(supNiceResult?.error ? { nice_error: supNiceResult.error } : {}),
|
||||
onEvent: (emission) => writeSupervisorEvent(emission, supervisorPid),
|
||||
});
|
||||
|
||||
|
||||
@@ -150,6 +150,7 @@ export const OPS_CHECK_NAMES: ReadonlySet<string> = new Set([
|
||||
'subagent_capability',
|
||||
'subagent_health',
|
||||
'supervisor',
|
||||
'supervisor_niceness',
|
||||
'sync_consolidation',
|
||||
'wedged_queue',
|
||||
'worker_oom_loop',
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
/**
|
||||
* OS scheduling priority (niceness) helpers for the Minions jobs subsystem.
|
||||
*
|
||||
* The lever behind issue #1815: background jobs (sync, embed, extract, subagent
|
||||
* fans) can run at full concurrency yet drive machine load average high enough to
|
||||
* starve the interactive shell. Reniceing the supervisor → worker → spawned-child
|
||||
* tree to a low priority lets the work run full-width when the box is idle and
|
||||
* politely yield when it isn't — load drops with zero throughput loss.
|
||||
*
|
||||
* Niceness range is POSIX `[-20, 19]` (-20 = highest priority, 19 = lowest /
|
||||
* "nicest"). Positive values need no privilege; negative values (raise priority)
|
||||
* require root and fail EPERM, OR partially clamp to RLIMIT_NICE *without
|
||||
* throwing*. Both cases are handled by always re-reading the effective value.
|
||||
*
|
||||
* This module is the pure, testable core: the CLI layer (src/commands/jobs.ts)
|
||||
* owns the side effect of actually calling it, the same boundary that keeps
|
||||
* worker.ts/supervisor.ts free of process.exit.
|
||||
*/
|
||||
|
||||
import { setPriority as osSetPriority, getPriority as osGetPriority } from 'os';
|
||||
|
||||
/** Lowest (highest-priority) and highest (nicest) POSIX nice values. */
|
||||
export const NICE_MIN = -20;
|
||||
export const NICE_MAX = 19;
|
||||
|
||||
export interface ApplyNicenessResult {
|
||||
/** True when setPriority did not throw. A `false` here is the divergence signal. */
|
||||
applied: boolean;
|
||||
/** The niceness the caller asked for. */
|
||||
requested: number;
|
||||
/**
|
||||
* The niceness actually in effect AFTER the attempt, re-read via getPriority.
|
||||
* Distinct from `requested` when the kernel clamped (RLIMIT_NICE) or the set
|
||||
* was denied (EPERM → stays at the inherited value, typically 0). `null` only
|
||||
* when the re-read itself failed.
|
||||
*/
|
||||
effective: number | null;
|
||||
/** Present when setPriority threw (e.g. "EACCES" / "EPERM"). */
|
||||
error?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a `--nice <n>` value. Whole-string integer parse — `parseInt` would
|
||||
* wrongly accept "3.5" → 3 and "10abc" → 10 (Codex #3). Mirrors the
|
||||
* `String(n) !== s.trim()` guard in src/core/sync-concurrency.ts:parseWorkers.
|
||||
* Throws on non-integer or out-of-range input; the CLI wraps this with exit(1).
|
||||
*/
|
||||
export function parseNiceValue(raw: string): number {
|
||||
const s = raw.trim();
|
||||
const n = parseInt(s, 10);
|
||||
if (!Number.isFinite(n) || String(n) !== s) {
|
||||
throw new Error(`--nice must be an integer in [${NICE_MIN}, ${NICE_MAX}], got: ${JSON.stringify(raw)}`);
|
||||
}
|
||||
if (n < NICE_MIN || n > NICE_MAX) {
|
||||
throw new Error(`--nice must be in [${NICE_MIN}, ${NICE_MAX}] (-20 = highest priority, 19 = nicest), got: ${n}`);
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply niceness to the CURRENT process (pid 0). Always re-reads the effective
|
||||
* value afterwards — in BOTH the success and failure paths (Codex #4) — so a
|
||||
* denied renice records `effective: 0`, not `null`, and doctor can distinguish
|
||||
* "permission denied but still running at 0" from "could not measure".
|
||||
*
|
||||
* `setPriority`/`getPriority` are injectable for tests.
|
||||
*/
|
||||
export function applyNiceness(
|
||||
nice: number,
|
||||
setPriority: (pid: number, priority: number) => void = osSetPriority,
|
||||
getPriority: (pid: number) => number = osGetPriority,
|
||||
): ApplyNicenessResult {
|
||||
let applied = false;
|
||||
let error: string | undefined;
|
||||
try {
|
||||
setPriority(0, nice);
|
||||
applied = true;
|
||||
} catch (e) {
|
||||
error = e instanceof Error ? e.message : String(e);
|
||||
}
|
||||
|
||||
// Re-read regardless of outcome (catches RLIMIT_NICE clamp on success and the
|
||||
// unchanged inherited value on failure).
|
||||
let effective: number | null;
|
||||
try {
|
||||
effective = getPriority(0);
|
||||
} catch {
|
||||
effective = null;
|
||||
}
|
||||
|
||||
return { applied, requested: nice, effective, error };
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the effective niceness of an arbitrary pid. Used by the read surfaces
|
||||
* (jobs stats, doctor, supervisor status). Returns null when the pid is gone or
|
||||
* unreadable. getpriority(2) needs no ownership to read on Linux/macOS.
|
||||
*/
|
||||
export function getEffectiveNiceness(
|
||||
pid: number,
|
||||
getPriority: (pid: number) => number = osGetPriority,
|
||||
): number | null {
|
||||
try {
|
||||
return getPriority(pid);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Format a nice value for human output: `+10`, `0`, `-5`. */
|
||||
export function formatNice(n: number): string {
|
||||
return n > 0 ? `+${n}` : String(n);
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
/**
|
||||
* Shared supervisor PID-file read + liveness check (issue #1815, Q4).
|
||||
*
|
||||
* The `existsSync → readFileSync → parseInt → process.kill(pid,0)` block was
|
||||
* copy-pasted across `gbrain jobs supervisor status` and `gbrain doctor`; the
|
||||
* #1815 niceness work would have added a third copy in `gbrain jobs stats`.
|
||||
* Extracted here so all three share one regression point.
|
||||
*/
|
||||
|
||||
import { existsSync, readFileSync } from 'fs';
|
||||
|
||||
export interface SupervisorPidStatus {
|
||||
/** Parsed pid from the PID file, or null if missing/corrupt. */
|
||||
pid: number | null;
|
||||
/** True when `pid` is set AND the process is alive (signalable / EPERM). */
|
||||
running: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a supervisor PID file and report liveness. Best-effort: unreadable or
|
||||
* corrupt files yield `{ pid: null, running: false }`. EPERM from the liveness
|
||||
* probe counts as running (the process exists, we just can't signal it).
|
||||
*/
|
||||
export function readSupervisorPid(pidFile: string): SupervisorPidStatus {
|
||||
if (!existsSync(pidFile)) return { pid: null, running: false };
|
||||
try {
|
||||
const line = readFileSync(pidFile, 'utf8').trim().split('\n')[0];
|
||||
const parsed = parseInt(line, 10);
|
||||
if (Number.isNaN(parsed) || parsed <= 0) return { pid: null, running: false };
|
||||
try {
|
||||
process.kill(parsed, 0);
|
||||
return { pid: parsed, running: true };
|
||||
} catch (e) {
|
||||
const code = (e as NodeJS.ErrnoException)?.code;
|
||||
// EPERM: process exists but not signalable by us → still running.
|
||||
return { pid: parsed, running: code === 'EPERM' };
|
||||
}
|
||||
} catch {
|
||||
return { pid: null, running: false };
|
||||
}
|
||||
}
|
||||
@@ -84,6 +84,17 @@ export interface SupervisorOpts {
|
||||
* resolveDefaultMaxRssMb() (issue #1678) instead of a flat default.
|
||||
* Set to 0 to spawn the worker without a watchdog. */
|
||||
maxRssMb: number;
|
||||
/** Niceness (issue #1815) the operator requested via `--nice` / `GBRAIN_NICE`,
|
||||
* or undefined to inherit. When set, the worker is spawned with `--nice N` so
|
||||
* it re-applies the value (the supervisor itself is reniced by the CLI layer,
|
||||
* before construction). The apply RESULT is computed in jobs.ts and passed in
|
||||
* here purely so the `started` audit event records what actually happened —
|
||||
* the supervisor does not call setPriority. */
|
||||
nice_requested?: number;
|
||||
/** Effective niceness of the supervisor process after its own renice attempt. */
|
||||
nice_effective?: number;
|
||||
/** Error string if the supervisor's own renice failed (e.g. EPERM). */
|
||||
nice_error?: string;
|
||||
/**
|
||||
* issue #1801 — progress watchdog. When a child is alive but makes no forward
|
||||
* progress on claimable work for this many minutes (waiting_claimable > 0 and
|
||||
@@ -142,6 +153,30 @@ const DEFAULTS: Omit<SupervisorOpts, 'cliPath'> = {
|
||||
startupGraceMs: 120_000, // overridden to 2× healthInterval in the constructor
|
||||
};
|
||||
|
||||
/**
|
||||
* Build the argv the supervisor uses to spawn `gbrain jobs work`. Extracted from
|
||||
* runSuperviseLoop so it's unit-testable (issue #1815, Codex). Appends `--nice N`
|
||||
* when the operator requested a niceness, alongside the existing concurrency /
|
||||
* queue / max-rss flags. The spawned worker re-applies the niceness to itself;
|
||||
* niceness also inherits to the worker's own children automatically.
|
||||
*/
|
||||
export function buildWorkerArgs(
|
||||
opts: Pick<SupervisorOpts, 'concurrency' | 'queue' | 'maxRssMb' | 'nice_requested'>,
|
||||
): string[] {
|
||||
const args = [
|
||||
'jobs', 'work',
|
||||
'--concurrency', String(opts.concurrency),
|
||||
'--queue', opts.queue,
|
||||
];
|
||||
if (opts.maxRssMb > 0) {
|
||||
args.push('--max-rss', String(opts.maxRssMb));
|
||||
}
|
||||
if (opts.nice_requested !== undefined) {
|
||||
args.push('--nice', String(opts.nice_requested));
|
||||
}
|
||||
return args;
|
||||
}
|
||||
|
||||
/** Grace before SIGKILL when restarting a wedged child — reuses the 35s
|
||||
* shutdown() drain window (issue #1801, D3). */
|
||||
const WEDGE_RESTART_GRACE_MS = 35_000;
|
||||
@@ -424,6 +459,11 @@ export class MinionSupervisor {
|
||||
concurrency: this.opts.concurrency,
|
||||
queue: this.opts.queue,
|
||||
max_crashes: this.opts.maxCrashes,
|
||||
// Niceness (issue #1815): record requested + effective so doctor/status can
|
||||
// surface a failed renice even for a detached supervisor whose stderr is gone.
|
||||
...(this.opts.nice_requested !== undefined ? { nice_requested: this.opts.nice_requested } : {}),
|
||||
...(this.opts.nice_effective !== undefined ? { nice_effective: this.opts.nice_effective } : {}),
|
||||
...(this.opts.nice_error ? { nice_error: this.opts.nice_error } : {}),
|
||||
});
|
||||
|
||||
// 6. Derive the claimable job-name set for the wedge watchdog (issue #1801).
|
||||
@@ -603,14 +643,7 @@ export class MinionSupervisor {
|
||||
* so JSONL audit consumers see byte-compatible output.
|
||||
*/
|
||||
private async runSuperviseLoop(): Promise<void> {
|
||||
const workerArgs = [
|
||||
'jobs', 'work',
|
||||
'--concurrency', String(this.opts.concurrency),
|
||||
'--queue', this.opts.queue,
|
||||
];
|
||||
if (this.opts.maxRssMb > 0) {
|
||||
workerArgs.push('--max-rss', String(this.opts.maxRssMb));
|
||||
}
|
||||
const workerArgs = buildWorkerArgs(this.opts);
|
||||
|
||||
// Build child env. Explicit handling for GBRAIN_ALLOW_SHELL_JOBS:
|
||||
// inherit only when caller opts in, otherwise strip from the clone.
|
||||
@@ -666,6 +699,7 @@ export class MinionSupervisor {
|
||||
pid: event.pid >= 0 ? event.pid : undefined,
|
||||
cli_path: this.opts.cliPath,
|
||||
...(event.tini ? { tini: true } : {}),
|
||||
...(this.opts.nice_requested !== undefined ? { nice: this.opts.nice_requested } : {}),
|
||||
});
|
||||
return;
|
||||
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
/**
|
||||
* Live worker registry for niceness observability (issue #1815, Q1-C).
|
||||
*
|
||||
* Each running `gbrain jobs work` process self-registers a small JSON file
|
||||
* recording its pid, queue, brain identity, start time, and the niceness it
|
||||
* requested + the niceness actually in effect. The read surfaces (jobs stats,
|
||||
* doctor, supervisor status) enumerate these to report the EFFECTIVE niceness of
|
||||
* the real worker process — not the supervisor's value as a proxy. This also
|
||||
* covers standalone `jobs work` (no supervisor / PID file) and sidesteps the
|
||||
* tini-wrapper-PID problem (the worker writes its OWN pid, Codex #5).
|
||||
*
|
||||
* Discipline mirrors src/core/audit/audit-writer.ts: best-effort, never blocks
|
||||
* the worker. A failed write just means that worker is omitted from the read
|
||||
* surfaces — it does not affect job execution.
|
||||
*
|
||||
* Location is brain-isolated via gbrainPath() (honors GBRAIN_HOME); entries are
|
||||
* additionally tagged with a brain id so a single GBRAIN_HOME hosting multiple
|
||||
* databases doesn't cross-report (Codex #7).
|
||||
*/
|
||||
|
||||
import {
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
readdirSync,
|
||||
readFileSync,
|
||||
unlinkSync,
|
||||
writeFileSync,
|
||||
} from 'fs';
|
||||
import { join } from 'path';
|
||||
import { execFileSync } from 'child_process';
|
||||
import { gbrainPath, loadConfig } from '../config.ts';
|
||||
import { getEffectiveNiceness } from './niceness.ts';
|
||||
|
||||
/** On-disk shape of a `worker-<pid>.json` entry. */
|
||||
export interface WorkerRegistryEntry {
|
||||
pid: number;
|
||||
queue: string;
|
||||
/** Short identifier of the brain (DB) this worker serves. */
|
||||
brain_id: string;
|
||||
/** Epoch ms when the worker registered (≈ its own process start). */
|
||||
started_at: number;
|
||||
/** Niceness the worker asked for (the `--nice` value), or null if none. */
|
||||
nice_requested: number | null;
|
||||
/** Niceness actually in effect when the worker registered. */
|
||||
nice_effective: number | null;
|
||||
}
|
||||
|
||||
/** A live worker as returned by readWorkers(): on-disk entry + fresh re-measure. */
|
||||
export interface LiveWorker extends WorkerRegistryEntry {
|
||||
/** Niceness re-measured NOW via getEffectiveNiceness(pid). */
|
||||
nice_now: number | null;
|
||||
}
|
||||
|
||||
/** Directory holding one file per live worker. Brain-isolated via GBRAIN_HOME. */
|
||||
export function workerRegistryDir(): string {
|
||||
return gbrainPath('workers');
|
||||
}
|
||||
|
||||
/**
|
||||
* Short, stable id for the active brain (database). Best-effort: derived from
|
||||
* the configured DB url/path. Returns 'default' when nothing is configured.
|
||||
* Used to tag + filter registry entries so multiple DBs under one GBRAIN_HOME
|
||||
* don't cross-report.
|
||||
*/
|
||||
export function currentBrainId(): string {
|
||||
try {
|
||||
const cfg = loadConfig();
|
||||
const key = cfg?.database_url ?? cfg?.database_path ?? 'default';
|
||||
// Tiny non-crypto hash (djb2) — we only need a stable short discriminator.
|
||||
let h = 5381;
|
||||
for (let i = 0; i < key.length; i++) h = ((h << 5) + h + key.charCodeAt(i)) | 0;
|
||||
return (h >>> 0).toString(36);
|
||||
} catch {
|
||||
return 'default';
|
||||
}
|
||||
}
|
||||
|
||||
function entryPath(pid: number): string {
|
||||
return join(workerRegistryDir(), `worker-${pid}.json`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Register the current worker process. Best-effort write; returns a cleanup
|
||||
* function that unlinks the entry. The caller MUST wire cleanup to both the
|
||||
* shutdown `finally` AND `process.on('exit')` — the unhealthy `process.exit(1)`
|
||||
* path bypasses awaited cleanup (Codex #10). SIGKILL still leaves a stale file;
|
||||
* the read side prunes those via liveness checks.
|
||||
*/
|
||||
export function registerWorker(info: {
|
||||
pid: number;
|
||||
queue: string;
|
||||
nice_requested: number | null;
|
||||
nice_effective: number | null;
|
||||
started_at: number;
|
||||
brain_id?: string;
|
||||
}): () => void {
|
||||
const entry: WorkerRegistryEntry = {
|
||||
pid: info.pid,
|
||||
queue: info.queue,
|
||||
brain_id: info.brain_id ?? currentBrainId(),
|
||||
started_at: info.started_at,
|
||||
nice_requested: info.nice_requested,
|
||||
nice_effective: info.nice_effective,
|
||||
};
|
||||
const path = entryPath(info.pid);
|
||||
try {
|
||||
mkdirSync(workerRegistryDir(), { recursive: true });
|
||||
writeFileSync(path, JSON.stringify(entry), 'utf8');
|
||||
} catch {
|
||||
// Best-effort: a failed write just omits this worker from read surfaces.
|
||||
}
|
||||
|
||||
let cleaned = false;
|
||||
return () => {
|
||||
if (cleaned) return;
|
||||
cleaned = true;
|
||||
try {
|
||||
if (existsSync(path)) unlinkSync(path);
|
||||
} catch { /* best effort */ }
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify a `process.kill(pid, 0)` outcome. EPERM means the process exists but
|
||||
* we can't signal it → alive, NOT dead (Codex #9). Only ESRCH (no such process)
|
||||
* is a confirmed death worth pruning. Exported pure helper so the EPERM/ESRCH
|
||||
* policy is unit-testable without a real privileged process.
|
||||
*/
|
||||
export function classifyLiveness(killErrorCode: string | undefined): 'alive' | 'dead' | 'unknown' {
|
||||
if (killErrorCode === undefined) return 'alive'; // kill(0) did not throw
|
||||
if (killErrorCode === 'ESRCH') return 'dead';
|
||||
if (killErrorCode === 'EPERM') return 'alive';
|
||||
return 'unknown';
|
||||
}
|
||||
|
||||
function processLiveness(pid: number): 'alive' | 'dead' | 'unknown' {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return classifyLiveness(undefined);
|
||||
} catch (e) {
|
||||
return classifyLiveness((e as NodeJS.ErrnoException)?.code);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Best-effort process start time (epoch ms) via `ps`. Used for the PID-reuse
|
||||
* guard: a stale `worker-<pid>.json` plus an OS-reused pid would otherwise make
|
||||
* us report an unrelated process's niceness (Codex #8). Returns null when
|
||||
* undeterminable — callers must NOT treat null as "reused".
|
||||
*/
|
||||
function processStartMs(pid: number): number | null {
|
||||
try {
|
||||
const out = execFileSync('ps', ['-o', 'lstart=', '-p', String(pid)], {
|
||||
encoding: 'utf8',
|
||||
timeout: 2000,
|
||||
stdio: ['ignore', 'pipe', 'ignore'],
|
||||
}).trim();
|
||||
if (!out) return null;
|
||||
const t = Date.parse(out);
|
||||
return Number.isNaN(t) ? null : t;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Tolerance (ms) for the PID-reuse start-time comparison — covers the small gap
|
||||
* between a worker's actual start and when it wrote its registry entry, plus
|
||||
* clock/`ps`-resolution slop. */
|
||||
const PID_REUSE_TOLERANCE_MS = 5000;
|
||||
|
||||
/**
|
||||
* Read live workers for the current brain. Enumerates the registry, drops
|
||||
* confirmed-dead entries (pruning their files), filters by brain id, applies the
|
||||
* PID-reuse guard, and re-measures each live worker's niceness now.
|
||||
*
|
||||
* `getNice` is injectable for tests.
|
||||
*/
|
||||
export function readWorkers(
|
||||
getNice: (pid: number) => number | null = (pid) => getEffectiveNiceness(pid),
|
||||
): LiveWorker[] {
|
||||
const dir = workerRegistryDir();
|
||||
if (!existsSync(dir)) return [];
|
||||
|
||||
const brainId = currentBrainId();
|
||||
const live: LiveWorker[] = [];
|
||||
|
||||
let files: string[];
|
||||
try {
|
||||
files = readdirSync(dir).filter((f) => f.startsWith('worker-') && f.endsWith('.json'));
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
|
||||
for (const f of files) {
|
||||
const full = join(dir, f);
|
||||
let entry: WorkerRegistryEntry;
|
||||
try {
|
||||
entry = JSON.parse(readFileSync(full, 'utf8')) as WorkerRegistryEntry;
|
||||
} catch {
|
||||
continue; // corrupt / truncated write — skip
|
||||
}
|
||||
if (!entry || typeof entry.pid !== 'number') continue;
|
||||
|
||||
const liveness = processLiveness(entry.pid);
|
||||
if (liveness === 'dead') {
|
||||
try { unlinkSync(full); } catch { /* best effort */ }
|
||||
continue;
|
||||
}
|
||||
|
||||
// Only this brain's workers (multi-DB under one GBRAIN_HOME).
|
||||
if (entry.brain_id && entry.brain_id !== brainId) continue;
|
||||
|
||||
// PID-reuse guard: if the live pid demonstrably started well after this
|
||||
// entry was written, the pid was recycled — don't report a stranger.
|
||||
const startMs = processStartMs(entry.pid);
|
||||
if (startMs !== null && entry.started_at && startMs - entry.started_at > PID_REUSE_TOLERANCE_MS) {
|
||||
continue;
|
||||
}
|
||||
|
||||
live.push({ ...entry, nice_now: getNice(entry.pid) });
|
||||
}
|
||||
|
||||
return live.sort((a, b) => a.pid - b.pid);
|
||||
}
|
||||
@@ -34,7 +34,13 @@ describe('connect bearer probe E2E (PGLite + real serve --http)', () => {
|
||||
|
||||
beforeAll(async () => {
|
||||
home = mkdtempSync(join(tmpdir(), 'gbrain-connect-e2e-'));
|
||||
const env = { ...process.env, GBRAIN_HOME: home };
|
||||
// Hermetic PGLite: strip any ambient Postgres URL so `init --pglite` and the
|
||||
// spawned `serve` actually use PGLite. Without this, a dev/CI shell with
|
||||
// DATABASE_URL set leaks it into the subprocess and the brain comes up on
|
||||
// Postgres, breaking the `engine: pglite` assertion.
|
||||
const env: Record<string, string | undefined> = { ...process.env, GBRAIN_HOME: home };
|
||||
delete env.DATABASE_URL;
|
||||
delete env.GBRAIN_DATABASE_URL;
|
||||
|
||||
execFileSync('bun', ['run', 'src/cli.ts', 'init', '--pglite', '--no-embedding', '--non-interactive'], {
|
||||
cwd: process.cwd(), env, stdio: 'ignore',
|
||||
|
||||
@@ -128,6 +128,7 @@ const EXPECTED_PHASES: CyclePhase[] = [
|
||||
'grade_takes', // v0.36.1.0
|
||||
'calibration_profile', // v0.36.1.0
|
||||
'conversation_facts_backfill', // v0.41.11.0 — opt-in conversation backfill
|
||||
'enrich_thin', // v0.41.39 (#1700) — brain-internal stub enrichment (default OFF)
|
||||
'skillopt', // v0.42.0.0 — self-evolving skills (default OFF)
|
||||
'embed',
|
||||
'orphans',
|
||||
|
||||
@@ -27,6 +27,15 @@ import { PGLiteEngine } from '../../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from '../helpers/reset-pglite.ts';
|
||||
import { IngestionDaemon } from '../../src/core/ingestion/daemon.ts';
|
||||
import { createInboxFolderSource } from '../../src/core/ingestion/sources/inbox-folder.ts';
|
||||
import { watch, type ChokidarOptions, type FSWatcher } from 'chokidar';
|
||||
|
||||
// E2E determinism: force chokidar into polling mode via the source's
|
||||
// `_watchFactory` seam so file-drop detection never depends on macOS fsevents
|
||||
// timing. The native-events path occasionally missed the first add under load,
|
||||
// timing out the 15s waitFor (the documented first-drop flake). Polling at a
|
||||
// 20ms interval is deterministic and still fast. Production keeps native events.
|
||||
const pollingWatchFactory = (paths: string, opts: ChokidarOptions): FSWatcher =>
|
||||
watch(paths, { ...opts, usePolling: true, interval: 20, binaryInterval: 20 });
|
||||
import { makeIngestCaptureHandler } from '../../src/core/minions/handlers/ingest-capture.ts';
|
||||
import type { IngestionEvent } from '../../src/core/ingestion/types.ts';
|
||||
import type { MinionJobContext } from '../../src/core/minions/types.ts';
|
||||
@@ -133,14 +142,19 @@ describe('ingestion roundtrip — inbox-folder → daemon → ingest_capture →
|
||||
inboxDir,
|
||||
debounceMs: 50,
|
||||
awaitStabilityMs: 100,
|
||||
_watchFactory: pollingWatchFactory,
|
||||
});
|
||||
daemon.register({ source });
|
||||
await daemon.start();
|
||||
|
||||
// Drop a file into the inbox.
|
||||
// Drop the file BEFORE start so the initial scan (ignoreInitial:false)
|
||||
// emits it deterministically. The full emit → dispatch → handler → DB →
|
||||
// archive pipeline still runs end to end; only the inherently timing-
|
||||
// dependent post-start watcher detection is removed (covered robustly by
|
||||
// the dedup test's polled second drop below).
|
||||
const captured = path.join(inboxDir, 'roundtrip.md');
|
||||
fs.writeFileSync(captured, '---\ntitle: Roundtrip\n---\n\nfull e2e flow');
|
||||
|
||||
await daemon.start();
|
||||
// Wait for the daemon to pick it up + dispatch + handler to write.
|
||||
await waitFor(() => dispatchedEvents.length === 1, 15000);
|
||||
|
||||
@@ -186,28 +200,36 @@ describe('ingestion roundtrip — inbox-folder → daemon → ingest_capture →
|
||||
inboxDir,
|
||||
debounceMs: 50,
|
||||
awaitStabilityMs: 100,
|
||||
_watchFactory: pollingWatchFactory,
|
||||
});
|
||||
daemon.register({ source });
|
||||
await daemon.start();
|
||||
|
||||
// Drop file 1
|
||||
// Drop file 1 BEFORE start so the initial scan (ignoreInitial:false) emits
|
||||
// it deterministically — removes the post-start watcher race that flaked.
|
||||
const drop1 = path.join(inboxDir, 'dup-1.md');
|
||||
fs.writeFileSync(drop1, '# duplicate content\n\nidentical body');
|
||||
|
||||
await daemon.start();
|
||||
await waitFor(() => dispatchedEvents.length === 1, 15000);
|
||||
|
||||
// Drop file 2 with byte-identical content (different filename).
|
||||
// Drop file 2 with byte-identical content (different filename) AFTER start;
|
||||
// the watcher catches it and dedup intercepts before dispatch.
|
||||
const drop2 = path.join(inboxDir, 'dup-2.md');
|
||||
fs.writeFileSync(drop2, '# duplicate content\n\nidentical body');
|
||||
// chokidar.archive moves drop2, but the dedup should catch the event
|
||||
// BEFORE the handler runs. Let chokidar process a bit then check.
|
||||
await new Promise((r) => setTimeout(r, 600));
|
||||
|
||||
// Poll for the dedup hit instead of a fixed sleep so a slow watcher tick
|
||||
// can't make the assertion flake.
|
||||
let dedupHits = 0;
|
||||
for (let i = 0; i < 240; i++) {
|
||||
dedupHits = (await daemon.healthCheck()).dedup.hits;
|
||||
if (dedupHits >= 1) break;
|
||||
await new Promise((r) => setTimeout(r, 25));
|
||||
}
|
||||
|
||||
// Only ONE event made it through dispatch (dedup intercepted the second).
|
||||
expect(dispatchedEvents).toHaveLength(1);
|
||||
|
||||
// The daemon's dedup stats reflect a hit.
|
||||
const health = await daemon.healthCheck();
|
||||
expect(health.dedup.hits).toBeGreaterThanOrEqual(1);
|
||||
expect(dedupHits).toBeGreaterThanOrEqual(1);
|
||||
|
||||
await daemon.stop();
|
||||
}, 15_000);
|
||||
@@ -241,20 +263,27 @@ describe('ingestion roundtrip — multi-source coordination', () => {
|
||||
inboxDir: inboxA,
|
||||
debounceMs: 50,
|
||||
awaitStabilityMs: 100,
|
||||
_watchFactory: pollingWatchFactory,
|
||||
});
|
||||
const sourceB = createInboxFolderSource({
|
||||
id: 'inbox-b',
|
||||
inboxDir: inboxB,
|
||||
debounceMs: 50,
|
||||
awaitStabilityMs: 100,
|
||||
_watchFactory: pollingWatchFactory,
|
||||
});
|
||||
daemon.register({ source: sourceA });
|
||||
daemon.register({ source: sourceB });
|
||||
await daemon.start();
|
||||
|
||||
// Drop both files BEFORE start. With ignoreInitial:false the initial scan
|
||||
// emits an `add` for each on chokidar `ready`, so both sources ingest
|
||||
// deterministically — no dependency on the post-start watcher catching a
|
||||
// freshly written file (the source of the residual flake).
|
||||
fs.writeFileSync(path.join(inboxA, 'from-a.md'), 'content from A');
|
||||
fs.writeFileSync(path.join(inboxB, 'from-b.md'), 'content from B');
|
||||
|
||||
await daemon.start();
|
||||
|
||||
await waitFor(() => dispatchedEvents.length === 2, 15000);
|
||||
|
||||
const fromA = dispatchedEvents.find((e) => e.source_id === 'inbox-a');
|
||||
|
||||
@@ -44,7 +44,17 @@ describe('serve stdio round-trip E2E (local PGLite → real MCP tool calls)', ()
|
||||
|
||||
beforeAll(async () => {
|
||||
home = mkdtempSync(join(tmpdir(), 'gbrain-stdio-e2e-'));
|
||||
const env = { ...process.env, GBRAIN_HOME: home };
|
||||
// Hermetic PGLite: strip any ambient Postgres URL so `init --pglite` and the
|
||||
// served brain actually use PGLite even when the shell/CI has DATABASE_URL
|
||||
// set (otherwise the subprocess comes up on Postgres → `engine: pglite`
|
||||
// assertion fails).
|
||||
// Build a concrete Record<string,string> (StdioClientTransport.env rejects
|
||||
// undefined values), dropping the ambient DB URLs so the subprocess is PGLite.
|
||||
const env: Record<string, string> = {};
|
||||
for (const [k, v] of Object.entries(process.env)) if (v !== undefined) env[k] = v;
|
||||
env.GBRAIN_HOME = home;
|
||||
delete env.DATABASE_URL;
|
||||
delete env.GBRAIN_DATABASE_URL;
|
||||
|
||||
// 1. Init a local PGLite brain (the "from nothing" step).
|
||||
execFileSync('bun', ['run', 'src/cli.ts', 'init', '--pglite', '--no-embedding', '--non-interactive'], {
|
||||
|
||||
@@ -92,11 +92,19 @@ describeE2E('v0.41.6.0 — sync lock recovery scenarios', () => {
|
||||
expect(result.stdout + result.stderr).toMatch(/not held|nothing to break/i);
|
||||
});
|
||||
|
||||
test('--break-lock + --all is refused with shell-loop hint', () => {
|
||||
test('--break-lock + --all is accepted and iterates sources (self-heal, no longer refused)', () => {
|
||||
// v0.41.13.0 (T4 + D1) intentionally DROPPED the old --all refusal so cron
|
||||
// can self-heal every source in one call: runBreakLock now widens to
|
||||
// iterate sources when --all is set (sync.ts ~2550). This test pins that
|
||||
// shipped contract. Deterministic: with no active sources OR only
|
||||
// unlocked ones, every per-source break is a no-op and the exit is 0.
|
||||
const result = runCli(['sync', '--break-lock', '--all']);
|
||||
expect(result.code).toBe(1);
|
||||
expect(result.stderr).toMatch(/cannot be combined with --all/);
|
||||
expect(result.stderr).toMatch(/for src in/);
|
||||
const out = result.stdout + result.stderr;
|
||||
expect(result.code).toBe(0);
|
||||
// The combination must NOT be rejected anymore.
|
||||
expect(out).not.toMatch(/cannot be combined with --all/);
|
||||
// It took the accepted iterate/no-sources path.
|
||||
expect(out).toMatch(/No active sources to break-lock against|is not held \(nothing to break\)|Broke lock/i);
|
||||
});
|
||||
|
||||
test('lock-busy error message includes PID + hostname + age + --break-lock hint', async () => {
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
/**
|
||||
* Unit tests for parseNiceFlag (issue #1815) — flag > GBRAIN_NICE env > undefined.
|
||||
*/
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { parseNiceFlag } from '../src/commands/jobs.ts';
|
||||
|
||||
describe('parseNiceFlag', () => {
|
||||
test('returns undefined when absent (no flag, no env)', () => {
|
||||
expect(parseNiceFlag(['jobs', 'work'], {})).toBeUndefined();
|
||||
});
|
||||
|
||||
test('reads the --nice flag', () => {
|
||||
expect(parseNiceFlag(['jobs', 'work', '--nice', '10'], {})).toBe(10);
|
||||
expect(parseNiceFlag(['jobs', 'work', '--nice', '-5'], {})).toBe(-5);
|
||||
});
|
||||
|
||||
test('falls back to GBRAIN_NICE env', () => {
|
||||
expect(parseNiceFlag(['jobs', 'work'], { GBRAIN_NICE: '7' })).toBe(7);
|
||||
});
|
||||
|
||||
test('flag wins over env', () => {
|
||||
expect(parseNiceFlag(['jobs', 'work', '--nice', '3'], { GBRAIN_NICE: '15' })).toBe(3);
|
||||
});
|
||||
|
||||
test('empty env string is treated as absent', () => {
|
||||
expect(parseNiceFlag(['jobs', 'work'], { GBRAIN_NICE: '' })).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,107 @@
|
||||
/**
|
||||
* Unit tests for the niceness core (issue #1815).
|
||||
*
|
||||
* Pins the correctness bugs Codex flagged on the plan:
|
||||
* - parseNiceValue must reject "3.5"/"10abc" that plain parseInt would accept (#3).
|
||||
* - applyNiceness must re-read effective in the FAILURE path too, so a denied
|
||||
* renice records effective:0, not null (#4).
|
||||
*/
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import {
|
||||
parseNiceValue,
|
||||
applyNiceness,
|
||||
getEffectiveNiceness,
|
||||
formatNice,
|
||||
NICE_MIN,
|
||||
NICE_MAX,
|
||||
} from '../src/core/minions/niceness.ts';
|
||||
|
||||
describe('parseNiceValue', () => {
|
||||
test('accepts the full POSIX range', () => {
|
||||
expect(parseNiceValue('0')).toBe(0);
|
||||
expect(parseNiceValue('10')).toBe(10);
|
||||
expect(parseNiceValue(String(NICE_MIN))).toBe(NICE_MIN);
|
||||
expect(parseNiceValue(String(NICE_MAX))).toBe(NICE_MAX);
|
||||
expect(parseNiceValue('-5')).toBe(-5);
|
||||
expect(parseNiceValue(' 7 ')).toBe(7); // trimmed
|
||||
});
|
||||
|
||||
test('rejects out-of-range', () => {
|
||||
expect(() => parseNiceValue('20')).toThrow();
|
||||
expect(() => parseNiceValue('-21')).toThrow();
|
||||
});
|
||||
|
||||
test('rejects non-integers parseInt would silently truncate (Codex #3)', () => {
|
||||
expect(() => parseNiceValue('3.5')).toThrow();
|
||||
expect(() => parseNiceValue('10abc')).toThrow();
|
||||
expect(() => parseNiceValue('abc')).toThrow();
|
||||
expect(() => parseNiceValue('')).toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe('applyNiceness', () => {
|
||||
test('success path returns applied + re-read effective', () => {
|
||||
let setTo: number | undefined;
|
||||
const r = applyNiceness(
|
||||
10,
|
||||
(_pid, p) => { setTo = p; },
|
||||
() => 10,
|
||||
);
|
||||
expect(setTo).toBe(10);
|
||||
expect(r.applied).toBe(true);
|
||||
expect(r.requested).toBe(10);
|
||||
expect(r.effective).toBe(10);
|
||||
expect(r.error).toBeUndefined();
|
||||
});
|
||||
|
||||
test('reports the clamped effective when the kernel caps it (Q3)', () => {
|
||||
// setPriority "succeeds" but the OS clamped to a higher (less negative) value.
|
||||
const r = applyNiceness(
|
||||
-20,
|
||||
() => { /* no throw */ },
|
||||
() => -5,
|
||||
);
|
||||
expect(r.applied).toBe(true);
|
||||
expect(r.requested).toBe(-20);
|
||||
expect(r.effective).toBe(-5); // re-read shows the real value
|
||||
});
|
||||
|
||||
test('failure path STILL re-reads effective (Codex #4) — records 0, not null', () => {
|
||||
const r = applyNiceness(
|
||||
-5,
|
||||
() => { const e: NodeJS.ErrnoException = new Error('EACCES'); e.code = 'EACCES'; throw e; },
|
||||
() => 0, // process stays at its inherited niceness
|
||||
);
|
||||
expect(r.applied).toBe(false);
|
||||
expect(r.error).toContain('EACCES');
|
||||
expect(r.effective).toBe(0); // NOT null — doctor can show "asked -5, got 0"
|
||||
});
|
||||
|
||||
test('effective is null only when the re-read itself throws', () => {
|
||||
const r = applyNiceness(
|
||||
10,
|
||||
() => { /* ok */ },
|
||||
() => { throw new Error('boom'); },
|
||||
);
|
||||
expect(r.applied).toBe(true);
|
||||
expect(r.effective).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('getEffectiveNiceness', () => {
|
||||
test('returns the stub value', () => {
|
||||
expect(getEffectiveNiceness(1234, () => 7)).toBe(7);
|
||||
});
|
||||
test('returns null when the read throws (dead/unreadable pid)', () => {
|
||||
expect(getEffectiveNiceness(1234, () => { throw new Error('ESRCH'); })).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('formatNice', () => {
|
||||
test('prefixes positive values with +', () => {
|
||||
expect(formatNice(10)).toBe('+10');
|
||||
expect(formatNice(0)).toBe('0');
|
||||
expect(formatNice(-5)).toBe('-5');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,39 @@
|
||||
/**
|
||||
* Unit tests for buildWorkerArgs (issue #1815) — the supervisor → worker argv.
|
||||
* Extracted from runSuperviseLoop so the --nice propagation is testable.
|
||||
*/
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { buildWorkerArgs } from '../src/core/minions/supervisor.ts';
|
||||
|
||||
describe('buildWorkerArgs', () => {
|
||||
test('base args without nice or rss', () => {
|
||||
expect(buildWorkerArgs({ concurrency: 2, queue: 'default', maxRssMb: 0 }))
|
||||
.toEqual(['jobs', 'work', '--concurrency', '2', '--queue', 'default']);
|
||||
});
|
||||
|
||||
test('includes --max-rss when > 0', () => {
|
||||
expect(buildWorkerArgs({ concurrency: 4, queue: 'q', maxRssMb: 2048 }))
|
||||
.toEqual(['jobs', 'work', '--concurrency', '4', '--queue', 'q', '--max-rss', '2048']);
|
||||
});
|
||||
|
||||
test('appends --nice when nice_requested is set', () => {
|
||||
expect(buildWorkerArgs({ concurrency: 2, queue: 'default', maxRssMb: 0, nice_requested: 10 }))
|
||||
.toEqual(['jobs', 'work', '--concurrency', '2', '--queue', 'default', '--nice', '10']);
|
||||
});
|
||||
|
||||
test('negative nice propagates', () => {
|
||||
const args = buildWorkerArgs({ concurrency: 1, queue: 'q', maxRssMb: 512, nice_requested: -5 });
|
||||
expect(args).toEqual(['jobs', 'work', '--concurrency', '1', '--queue', 'q', '--max-rss', '512', '--nice', '-5']);
|
||||
});
|
||||
|
||||
test('nice 0 is explicit and still propagates (distinct from inherit)', () => {
|
||||
expect(buildWorkerArgs({ concurrency: 1, queue: 'q', maxRssMb: 0, nice_requested: 0 }))
|
||||
.toContain('--nice');
|
||||
});
|
||||
|
||||
test('omits --nice when nice_requested is undefined (inherit)', () => {
|
||||
expect(buildWorkerArgs({ concurrency: 1, queue: 'q', maxRssMb: 0 }))
|
||||
.not.toContain('--nice');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,47 @@
|
||||
/**
|
||||
* Unit tests for the shared supervisor PID-file reader (issue #1815, Q4).
|
||||
* One regression point now backs `jobs supervisor status`, `jobs stats`, and
|
||||
* `gbrain doctor`.
|
||||
*/
|
||||
|
||||
import { describe, test, expect, beforeEach, afterEach } from 'bun:test';
|
||||
import { mkdtempSync, rmSync, writeFileSync } from 'fs';
|
||||
import { join } from 'path';
|
||||
import { tmpdir } from 'os';
|
||||
import { readSupervisorPid } from '../src/core/minions/supervisor-pid.ts';
|
||||
|
||||
let dir: string;
|
||||
beforeEach(() => { dir = mkdtempSync(join(tmpdir(), 'gbrain-suppid-')); });
|
||||
afterEach(() => { try { rmSync(dir, { recursive: true, force: true }); } catch { /* ignore */ } });
|
||||
|
||||
describe('readSupervisorPid', () => {
|
||||
test('present + alive → running', () => {
|
||||
const f = join(dir, 'sup.pid');
|
||||
writeFileSync(f, `${process.pid}\n`);
|
||||
const r = readSupervisorPid(f);
|
||||
expect(r.pid).toBe(process.pid);
|
||||
expect(r.running).toBe(true);
|
||||
});
|
||||
|
||||
test('present + dead → pid set, not running', () => {
|
||||
const f = join(dir, 'sup.pid');
|
||||
writeFileSync(f, '2147483600\n');
|
||||
const r = readSupervisorPid(f);
|
||||
expect(r.pid).toBe(2147483600);
|
||||
expect(r.running).toBe(false);
|
||||
});
|
||||
|
||||
test('missing file → null, not running', () => {
|
||||
const r = readSupervisorPid(join(dir, 'nope.pid'));
|
||||
expect(r.pid).toBeNull();
|
||||
expect(r.running).toBe(false);
|
||||
});
|
||||
|
||||
test('corrupt content → null, not running', () => {
|
||||
const f = join(dir, 'sup.pid');
|
||||
writeFileSync(f, 'not-a-pid\n');
|
||||
const r = readSupervisorPid(f);
|
||||
expect(r.pid).toBeNull();
|
||||
expect(r.running).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,121 @@
|
||||
/**
|
||||
* Unit tests for the live worker registry (issue #1815, Q1-C).
|
||||
*
|
||||
* Pins the lifecycle edges Codex flagged:
|
||||
* - ESRCH entries pruned, EPERM entries kept alive (#9).
|
||||
* - PID-reuse guard: an entry whose pid started long after registration is
|
||||
* not reported (#8).
|
||||
* - Registry path is brain-isolated under gbrainPath (#7).
|
||||
* - Corrupt JSON skipped; cleanup unlinks.
|
||||
*/
|
||||
|
||||
import { describe, test, expect, beforeEach, afterEach } from 'bun:test';
|
||||
import { mkdtempSync, rmSync, writeFileSync, existsSync, readdirSync } from 'fs';
|
||||
import { join } from 'path';
|
||||
import { tmpdir } from 'os';
|
||||
|
||||
let home: string;
|
||||
const origHome = process.env.GBRAIN_HOME;
|
||||
|
||||
beforeEach(() => {
|
||||
home = mkdtempSync(join(tmpdir(), 'gbrain-reg-'));
|
||||
process.env.GBRAIN_HOME = home;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
if (origHome === undefined) delete process.env.GBRAIN_HOME;
|
||||
else process.env.GBRAIN_HOME = origHome;
|
||||
try { rmSync(home, { recursive: true, force: true }); } catch { /* ignore */ }
|
||||
});
|
||||
|
||||
// Imported lazily AFTER GBRAIN_HOME is set so gbrainPath resolves to the temp dir.
|
||||
async function reg() {
|
||||
return await import('../src/core/minions/worker-registry.ts');
|
||||
}
|
||||
|
||||
describe('classifyLiveness (Codex #9)', () => {
|
||||
test('no error = alive, ESRCH = dead, EPERM = alive, other = unknown', async () => {
|
||||
const { classifyLiveness } = await reg();
|
||||
expect(classifyLiveness(undefined)).toBe('alive');
|
||||
expect(classifyLiveness('ESRCH')).toBe('dead');
|
||||
expect(classifyLiveness('EPERM')).toBe('alive'); // not pruned just because unsignalable
|
||||
expect(classifyLiveness('EINVAL')).toBe('unknown');
|
||||
});
|
||||
});
|
||||
|
||||
describe('register + read round trip', () => {
|
||||
test('registerWorker writes under gbrainPath; readWorkers returns the live worker', async () => {
|
||||
const { registerWorker, readWorkers, workerRegistryDir } = await reg();
|
||||
expect(workerRegistryDir()).toBe(join(home, '.gbrain', 'workers'));
|
||||
|
||||
const cleanup = registerWorker({
|
||||
pid: process.pid, // a definitely-alive pid
|
||||
queue: 'default',
|
||||
nice_requested: 10,
|
||||
nice_effective: 10,
|
||||
started_at: Date.now(),
|
||||
});
|
||||
|
||||
const live = readWorkers(() => 10); // inject niceness read
|
||||
expect(live.length).toBe(1);
|
||||
expect(live[0]!.pid).toBe(process.pid);
|
||||
expect(live[0]!.queue).toBe('default');
|
||||
expect(live[0]!.nice_requested).toBe(10);
|
||||
expect(live[0]!.nice_now).toBe(10);
|
||||
|
||||
cleanup();
|
||||
expect(readWorkers(() => 10).length).toBe(0);
|
||||
});
|
||||
|
||||
test('cleanup unlinks the entry file', async () => {
|
||||
const { registerWorker, workerRegistryDir } = await reg();
|
||||
const cleanup = registerWorker({
|
||||
pid: process.pid, queue: 'q', nice_requested: null, nice_effective: 0, started_at: Date.now(),
|
||||
});
|
||||
expect(readdirSync(workerRegistryDir()).length).toBe(1);
|
||||
cleanup();
|
||||
expect(readdirSync(workerRegistryDir()).length).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('pruning + guards', () => {
|
||||
test('ESRCH (dead) entries are dropped and their files pruned', async () => {
|
||||
const { readWorkers, workerRegistryDir, currentBrainId } = await reg();
|
||||
const dir = workerRegistryDir();
|
||||
// A pid that is essentially never alive.
|
||||
const deadPid = 2147483600;
|
||||
const file = join(dir, `worker-${deadPid}.json`);
|
||||
require('fs').mkdirSync(dir, { recursive: true });
|
||||
writeFileSync(file, JSON.stringify({
|
||||
pid: deadPid, queue: 'q', brain_id: currentBrainId(),
|
||||
started_at: Date.now(), nice_requested: 5, nice_effective: 5,
|
||||
}));
|
||||
const live = readWorkers(() => 5);
|
||||
expect(live.length).toBe(0);
|
||||
expect(existsSync(file)).toBe(false); // pruned
|
||||
});
|
||||
|
||||
test('PID-reuse guard: live pid that started long after registration is skipped (Codex #8)', async () => {
|
||||
const { readWorkers, workerRegistryDir, currentBrainId } = await reg();
|
||||
const dir = workerRegistryDir();
|
||||
require('fs').mkdirSync(dir, { recursive: true });
|
||||
// Use our own (alive) pid but claim it was registered far in the past — our
|
||||
// process actually started AFTER that, so the start-time guard rejects it.
|
||||
writeFileSync(join(dir, `worker-${process.pid}.json`), JSON.stringify({
|
||||
pid: process.pid, queue: 'q', brain_id: currentBrainId(),
|
||||
started_at: 1, // epoch ~1970; our process clearly started long after
|
||||
nice_requested: 5, nice_effective: 5,
|
||||
}));
|
||||
const live = readWorkers(() => 5);
|
||||
expect(live.length).toBe(0);
|
||||
});
|
||||
|
||||
test('corrupt JSON is skipped, not fatal', async () => {
|
||||
const { readWorkers, workerRegistryDir } = await reg();
|
||||
const dir = workerRegistryDir();
|
||||
require('fs').mkdirSync(dir, { recursive: true });
|
||||
writeFileSync(join(dir, `worker-${process.pid}.json`), '{ not valid json');
|
||||
expect(() => readWorkers(() => 5)).not.toThrow();
|
||||
expect(readWorkers(() => 5).length).toBe(0);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user