From f8682574057e851a8d4fe2ca90d1304a1dcf273b Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Wed, 3 Jun 2026 22:40:12 -0700 Subject: [PATCH] v0.42.24.0 fix(minions): route lock claim/renewLock through direct session pool (#1822) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(minions): route lock claim/renewLock through direct session pool The Minion lock heartbeat (claim + renewLock) ran every UPDATE through engine.executeRaw(), which is hardcoded to the read pool. On Supabase that is the transaction-mode pooler (6543), which recycles connections per transaction. A lock is held for minutes, so the pooler periodically reaps the socket mid-heartbeat -> CONNECTION_ENDED -> the lock looks expired -> the worker force-evicts its own job and the claim loop wedges silently. Add BrainEngine.executeRawDirect(): same contract as executeRaw, but routes to the direct session-mode pool (5432, GBRAIN_DIRECT_DATABASE_URL) when dual-pool is active. No-op delegation on PGLite / non-Supabase / kill-switch. claim/renewLock now use it. Single-statement UPDATEs only, so the double-claim guard and the renewLock no-inline-retry contract are preserved. Statements inside an open transaction keep their tx connection (in-transaction guard keys on peekReadPool() !== _sql); the lock hot-path never runs inside transaction(). The Postgres impl shares its cancellation plumbing with executeRaw via a private runUnsafe helper. New test/postgres-execute-raw-direct.test.ts covers the routing decision (dual-pool on/off x in-tx/not + abort short-circuit) without a live Postgres; queue-lock-retry.test.ts gains a guard that claim can never fall back to executeRaw. Co-Authored-By: Claude Opus 4.8 (1M context) * v0.42.24.0 chore: bump version and changelog Co-Authored-By: Claude Opus 4.8 (1M context) * docs: update project documentation for v0.42.24.0 Document executeRawDirect on the BrainEngine contract and the claim/renewLock direct-session-pool routing in KEY_FILES.md. Co-Authored-By: Claude Opus 4.8 (1M context) * test: make D3 executeRaw-no-retry guard refactor-aware The DRY refactor in this PR extracted executeRaw/executeRawDirect's shared cancellation plumbing into a private runUnsafe(conn, ...) helper, so the single conn.unsafe() call moved out of executeRaw's body. The D3 guard read executeRaw's source and asserted conn.unsafe( appeared exactly once there, which now fails (it's zero — executeRaw delegates). The D3 invariant (no per-call retry wrapper) is unchanged; it just spans the delegate now. Update the guard to check both public methods delegate to runUnsafe without reconnect/retry, and assert the exactly-once conn.unsafe + cancel-only catch in runUnsafe. Also extends coverage to executeRawDirect so the lock hot-path can't reintroduce a retry wrapper either. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 12 +++ TODOS.md | 21 +++++ VERSION | 2 +- docs/architecture/KEY_FILES.md | 4 +- package.json | 2 +- src/core/engine.ts | 18 ++++ src/core/minions/queue.ts | 10 +- src/core/pglite-engine.ts | 14 +++ src/core/postgres-engine.ts | 60 ++++++++++-- test/connection-resilience.test.ts | 50 ++++++---- test/postgres-execute-raw-direct.test.ts | 111 +++++++++++++++++++++++ test/queue-lock-retry.test.ts | 6 +- 12 files changed, 279 insertions(+), 31 deletions(-) create mode 100644 test/postgres-execute-raw-direct.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a36440213..1f9d14174 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ All notable changes to GBrain will be documented in this file. +## [0.42.24.0] - 2026-06-03 + +**Minion workers no longer silently wedge mid-job on a Supabase brain.** The background worker that runs your cron jobs, enrich fan-out, and autopilot cycle holds a lock on each job and heartbeats it every couple of seconds to say "still working." On a Supabase brain that heartbeat was running on the transaction-mode pooler (the high-traffic 6543 port), which recycles its connections per transaction. A lock is held open for minutes, so the pooler would periodically drop the socket mid-heartbeat. The worker read that dropped socket as "the lock expired," force-evicted its own in-flight job, and then sat in a claim loop holding nothing — process alive, no errors in the log, just quietly doing no work. It showed up most under heavy `enrich` load. + +The fix routes only the lock hot-path (`claim` and `renewLock`) to the direct **session-mode** pool (port 5432, `GBRAIN_DIRECT_DATABASE_URL`), which holds its connection open for the life of the worker so heartbeats survive. gbrain already shipped this dual-pool design for DDL and bulk work; the lock path just never used it. No new infrastructure — the direct URL was already in your config. + +- **Nothing to configure.** `gbrain upgrade` and the worker uses the right pool automatically on any Supabase brain. PGLite (the zero-config default) has no pooler, so this is a no-op there — same behavior on both engines. +- **Atomicity is preserved.** Statements inside an open transaction still run on the transaction's own connection; only the standalone lock heartbeat (which never runs inside a transaction) gets rerouted. There's a kill-switch (`GBRAIN_DISABLE_DIRECT_POOL`) if you ever need the old behavior. +- **Empirically:** a heartbeat survived 4/4 beats over 8s on the session pool, vs. connection-drop storms on the transaction pooler. + +### For contributors +New `BrainEngine.executeRawDirect()` — same contract as `executeRaw`, but routes to the direct pool when dual-pool is active (no-op delegation on PGLite / non-Supabase / kill-switch). `claim`/`renewLock` in `minions/queue.ts` point at it. The Postgres impl shares its cancellation plumbing with `executeRaw` via a private `runUnsafe` helper; the in-transaction guard keys on `peekReadPool() !== _sql` so a tx clone is detected and never rerouted. New `test/postgres-execute-raw-direct.test.ts` covers the routing decision (dual-pool on/off × in-tx/not, plus abort short-circuit) without a live Postgres; `test/queue-lock-retry.test.ts` gains a guard that `claim` can never fall back to `executeRaw`. Eng review cleared the plan; the four lock/pool guards stay green. ## [0.42.23.0] - 2026-06-03 **`gbrain jobs work` and `gbrain jobs supervisor` take a `--nice ` 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. diff --git a/TODOS.md b/TODOS.md index 0c4728cac..9d45c5d1a 100644 --- a/TODOS.md +++ b/TODOS.md @@ -95,6 +95,27 @@ structural change). Plan + GSTACK REVIEW REPORT at (`apply-edits.ts:311`) + `mkdirSync(recursive)` in `runOptimizationLoop` (`src/core/skillopt/orchestrator.ts`). Small, own PR. +## Minion-lock direct-pool follow-up (v0.42+) + +Filed from the eng-review of the lock-claim/renewLock → direct-session-pool fix +(PR #1816, now folded into `garrytan/minion-locks-session-pool`). Deliberately +scoped OUT of that change; not a regression. + +- [ ] **P3 — Size the direct session pool for enrich fan-out.** The lock + hot-path (`claim`/`renewLock`) now routes through the direct session-mode pool + (port 5432) via `executeRawDirect`. Supabase's session-mode pool has a far + smaller connection ceiling than the transaction pooler (6543). `executeRawDirect` + checks out per-statement (not held open), so the risk is bounded by *concurrent + in-flight heartbeats*, not duration — but under heavy `enrich` fan-out (many + Minion workers each heartbeating at once) the smaller pool could contend or + exhaust. **Why:** a starved session pool would reintroduce the exact wedge class + the fix removes, just from a different cause. **Current state:** direct pool size + comes from `resolveDirectPoolSize` / `DEFAULT_DIRECT_POOL_SIZE` + (`src/core/connection-manager.ts`); no fan-out-aware tuning. **Where to start:** + measure concurrent heartbeat count under a realistic `enrich` burst, compare to + `DEFAULT_DIRECT_POOL_SIZE`, and either raise the default or add a + worker-count-aware knob. **Depends on:** PR #1816 landing first. + ## v0.42.12.0 #1685 brain-health-as-solved follow-ups (v0.42+) Deferred from the v0.42.12.0 wave (issue #1685, the posture umbrella over #1678/#1735). diff --git a/VERSION b/VERSION index 7d13f161f..6997a4e12 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.42.23.0 \ No newline at end of file +0.42.24.0 \ No newline at end of file diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 830de90ff..f450d8a4a 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -9,7 +9,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this (`scripts/check-key-files-current-state.sh`). - `src/core/operations.ts` — Contract-first operation definitions (the foundation). Exports upload validators `validateUploadPath`, `validatePageSlug`, `validateFilename`, plus `matchesSlugAllowList(slug, prefixes)` (glob matcher: `/*` matches recursive children; bare `` matches exact only). `OperationContext.remote` is a REQUIRED field flagging untrusted callers; `OperationContext.allowedSlugPrefixes` is the trusted-workspace allow-list set by the dream cycle; `OperationContext.auth?: AuthInfo` is threaded through HTTP dispatch for scope enforcement in `serve-http.ts` before the op runs. `put_page` enforces: when `viaSubagent` and `allowedSlugPrefixes` is set, slug must match the allow-list; else the legacy `wiki/agents//...` namespace check applies. Auto-link skipped only when `remote=true && !trustedWorkspace`. Every `Operation` carries `scope?: 'read' | 'write' | 'admin'` + `localOnly?: boolean`; `sync_brain`, `file_upload`, `file_list`, `file_url` are `admin + localOnly` (rejected over HTTP). Four trust-boundary call sites (`put_page` allowlist, file_upload trust-narrowing, submit_job protected-name guard, auto-link skip) use FAIL-CLOSED semantics: `ctx.remote === false` for trusted-only sites, `ctx.remote !== false` for "untrust unless explicit-false" — anything not strictly `false` is treated as remote (closes the HTTP MCP shell-job RCE where a read+write OAuth token could submit `shell` jobs). `sourceScopeOpts(ctx)` encodes the source-scoped read precedence ladder — federated array (`ctx.auth.allowedSources`) wins over scalar (`ctx.sourceId`/`ctx.auth.sourceId`) over nothing; every read-side op handler routes through it so a source-bound OAuth client can't see neighboring sources via `search`/`query`/`list_pages`/`get_page`/`find_experts`/query's image path. `put_page`'s inline disk write-through is the shared `writePageThrough` helper (`src/core/write-through.ts`), ATOMIC via temp-sibling + rename so a crash or concurrent `gbrain sync` can't read a half-written `.md`; same helper backs `gbrain brainstorm/lsd --save`. -- `src/core/engine.ts` — Pluggable engine interface (BrainEngine). `clampSearchLimit(limit, default, cap)` takes an explicit cap so per-operation caps can be tighter than `MAX_SEARCH_LIMIT`. Exports `LinkBatchInput`/`TimelineBatchInput` for the bulk-insert API (`addLinksBatch`/`addTimelineEntriesBatch`). `readonly kind: 'postgres' | 'pglite'` discriminator lets `src/core/migrate.ts` and others branch without `instanceof` + dynamic imports. Methods: `batchLoadEmotionalInputs(slugs?)` (CTE-shaped read with per-table aggregates so page × N tags × M takes never produces N×M rows), `setEmotionalWeightBatch(rows)` (`UPDATE FROM unnest($1::text[],$2::text[],$3::real[])` composite-keyed on `(slug, source_id)`), `getRecentSalience(opts)`, `findAnomalies(opts)`. `PageFilters` has `sort?: 'updated_desc'|'updated_asc'|'created_desc'|'slug'` + `PAGE_SORT_SQL` whitelist consumed by both engines. `listAllPageRefs(): Promise>` ordered by `(source_id, slug)` — cheap cross-source enumeration replacing the `getAllSlugs()→getPage(slug)` N+1 (which silently defaulted to `source_id='default'`); parity across postgres-engine.ts + pglite-engine.ts; Pinned by `test/e2e/multi-source-bug-class.test.ts`. `SearchOpts`+`PageFilters` add `sourceIds?: string[]` (federated read axis; both engines apply `WHERE source_id = ANY($N::text[])` when set, preserve scalar `sourceId` fast path when unset); `traverseGraph(slug, depth, opts?)` and `traversePaths(slug, opts?)` accept `opts.sourceId`/`opts.sourceIds`. `traverseGraph` opts has `frontierCap?: number` (per-iteration recursive-CTE cap, approx per-BFS-layer); return type `Promise` for MCP wire stability; export `TraverseGraphOpts`; Postgres uses parenthesized `LIMIT N ORDER BY (slug, id)` inside the recursive term, PGLite mirrors with positional params; Pinned by `test/regressions/v0_36_frontier_cap.test.ts`. Phantom-redirect methods: `refreshPageBody(slug, sourceId, compiled_truth, timeline, content_hash)` narrow-UPDATEs three columns + updated_at, skipping soft-deleted rows (content_hash refresh required so `gbrain sync` sees the canonical as unchanged after fence merge); `migrateFactsToCanonical(phantomSlug, canonicalSlug, sourceId)` UPDATEs `entity_slug`+`source_markdown_slug` on every active fact row keyed on the phantom, preserving embedding/validUntil/kind/status/source_session/confidence; parity at `test/phantom-redirect-engine-parity.test.ts`. `getAdjacencyBoosts(pageIds): Promise>` powers the per-query graph-signals stage — one SQL query returning inbound-link counts among top-K plus a cross-source count (links from differing `source_id`); `COALESCE(p.source_id,'default')` null safety, `HAVING >= 1`, cross-source CASE-WHEN excludes the target's own source; parity SQL across both engines; `SearchResult` gains optional `base_score`, `backlink_boost`, `salience_boost`, `recency_boost`, `exact_match_boost`, `graph_adjacency_boost`, `graph_cross_source_boost`, `session_demote_factor`, `reranker_delta` + internal staging fields; Pinned by `test/e2e/graph-signals-engine.test.ts`. Two REQUIRED methods: `deletePages(slugs, {sourceId}): Promise` (single-batch primitive returning slugs actually deleted) and `resolveSlugsByPaths(paths, {sourceId}): Promise>` (batch path→slug lookup); `sourceId` REQUIRED on both at the type level (asymmetric with single-row `deletePage` which keeps optional/'default'); both short-circuit on empty input and throw when `> DELETE_BATCH_SIZE`. Embedding-signature stale-detection quartet: `countStaleChunks(opts?)` gains optional `signature?: string` widening the stale predicate from `embedding IS NULL` to ALSO include chunks whose JOINed page `embedding_signature IS NOT NULL AND <> $signature` (NULL signature is GRANDFATHERED, never counted; omit `signature` for the legacy NULL-only count); `sumStaleChunkChars(opts?: {sourceId?, signature?}): Promise` = `SUM(LENGTH(chunk_text))` over stale chunks (same predicate + embed_skip filter + optional sourceId scope), used by `gbrain sync --all` cost preview via `estimateCostFromChars`; `setPageEmbeddingSignature(slug, {sourceId?, signature})` stamps `pages.embedding_signature` after a page's chunks (re)embed, idempotent no-op when page absent; `invalidateStaleSignatureEmbeddings({signature, sourceId?}): Promise` NULLs `embedding`+`embedded_at` on every chunk whose page signature is set AND differs, returning the count, called BEFORE `listStaleChunks` so signature-drift pages flow through the NULL-embedding keyset cursor unchanged (NULL never invalidated). Widens `findOrphanPages(opts?: {sourceId?, sourceIds?})` (candidate-side scoping only; inbound links counted from any source). Pinned by `test/sum-stale-chunk-chars.test.ts`, `test/embedding-signature-stale.test.ts`, `test/e2e/engine-parity.test.ts`. Free-text alias layer: `resolveAliases(aliasNorms, opts?): Promise>>` (READ; maps each normalized alias to declaring `(slug, source_id)` pairs, source-scoped) and `setPageAliases(slug, sourceId, aliasNorms)` (WRITE; replaces the full alias set, delete-then-insert, empty clears, idempotent on the unique triple), called by the `importFromContent` ingest projection and the `reindex --aliases` backfill; parity across both engines, Pinned by `test/search/page-aliases-engine.test.ts`. `searchVector` in both engines injects the shared `buildBestPerPagePoolCte` per-page max-pool so a page surfaces on its strongest chunk. +- `src/core/engine.ts` — Pluggable engine interface (BrainEngine). `clampSearchLimit(limit, default, cap)` takes an explicit cap so per-operation caps can be tighter than `MAX_SEARCH_LIMIT`. Exports `LinkBatchInput`/`TimelineBatchInput` for the bulk-insert API (`addLinksBatch`/`addTimelineEntriesBatch`). `readonly kind: 'postgres' | 'pglite'` discriminator lets `src/core/migrate.ts` and others branch without `instanceof` + dynamic imports. Methods: `batchLoadEmotionalInputs(slugs?)` (CTE-shaped read with per-table aggregates so page × N tags × M takes never produces N×M rows), `setEmotionalWeightBatch(rows)` (`UPDATE FROM unnest($1::text[],$2::text[],$3::real[])` composite-keyed on `(slug, source_id)`), `getRecentSalience(opts)`, `findAnomalies(opts)`. `PageFilters` has `sort?: 'updated_desc'|'updated_asc'|'created_desc'|'slug'` + `PAGE_SORT_SQL` whitelist consumed by both engines. `listAllPageRefs(): Promise>` ordered by `(source_id, slug)` — cheap cross-source enumeration replacing the `getAllSlugs()→getPage(slug)` N+1 (which silently defaulted to `source_id='default'`); parity across postgres-engine.ts + pglite-engine.ts; Pinned by `test/e2e/multi-source-bug-class.test.ts`. `SearchOpts`+`PageFilters` add `sourceIds?: string[]` (federated read axis; both engines apply `WHERE source_id = ANY($N::text[])` when set, preserve scalar `sourceId` fast path when unset); `traverseGraph(slug, depth, opts?)` and `traversePaths(slug, opts?)` accept `opts.sourceId`/`opts.sourceIds`. `traverseGraph` opts has `frontierCap?: number` (per-iteration recursive-CTE cap, approx per-BFS-layer); return type `Promise` for MCP wire stability; export `TraverseGraphOpts`; Postgres uses parenthesized `LIMIT N ORDER BY (slug, id)` inside the recursive term, PGLite mirrors with positional params; Pinned by `test/regressions/v0_36_frontier_cap.test.ts`. Phantom-redirect methods: `refreshPageBody(slug, sourceId, compiled_truth, timeline, content_hash)` narrow-UPDATEs three columns + updated_at, skipping soft-deleted rows (content_hash refresh required so `gbrain sync` sees the canonical as unchanged after fence merge); `migrateFactsToCanonical(phantomSlug, canonicalSlug, sourceId)` UPDATEs `entity_slug`+`source_markdown_slug` on every active fact row keyed on the phantom, preserving embedding/validUntil/kind/status/source_session/confidence; parity at `test/phantom-redirect-engine-parity.test.ts`. `getAdjacencyBoosts(pageIds): Promise>` powers the per-query graph-signals stage — one SQL query returning inbound-link counts among top-K plus a cross-source count (links from differing `source_id`); `COALESCE(p.source_id,'default')` null safety, `HAVING >= 1`, cross-source CASE-WHEN excludes the target's own source; parity SQL across both engines; `SearchResult` gains optional `base_score`, `backlink_boost`, `salience_boost`, `recency_boost`, `exact_match_boost`, `graph_adjacency_boost`, `graph_cross_source_boost`, `session_demote_factor`, `reranker_delta` + internal staging fields; Pinned by `test/e2e/graph-signals-engine.test.ts`. Two REQUIRED methods: `deletePages(slugs, {sourceId}): Promise` (single-batch primitive returning slugs actually deleted) and `resolveSlugsByPaths(paths, {sourceId}): Promise>` (batch path→slug lookup); `sourceId` REQUIRED on both at the type level (asymmetric with single-row `deletePage` which keeps optional/'default'); both short-circuit on empty input and throw when `> DELETE_BATCH_SIZE`. Embedding-signature stale-detection quartet: `countStaleChunks(opts?)` gains optional `signature?: string` widening the stale predicate from `embedding IS NULL` to ALSO include chunks whose JOINed page `embedding_signature IS NOT NULL AND <> $signature` (NULL signature is GRANDFATHERED, never counted; omit `signature` for the legacy NULL-only count); `sumStaleChunkChars(opts?: {sourceId?, signature?}): Promise` = `SUM(LENGTH(chunk_text))` over stale chunks (same predicate + embed_skip filter + optional sourceId scope), used by `gbrain sync --all` cost preview via `estimateCostFromChars`; `setPageEmbeddingSignature(slug, {sourceId?, signature})` stamps `pages.embedding_signature` after a page's chunks (re)embed, idempotent no-op when page absent; `invalidateStaleSignatureEmbeddings({signature, sourceId?}): Promise` NULLs `embedding`+`embedded_at` on every chunk whose page signature is set AND differs, returning the count, called BEFORE `listStaleChunks` so signature-drift pages flow through the NULL-embedding keyset cursor unchanged (NULL never invalidated). Widens `findOrphanPages(opts?: {sourceId?, sourceIds?})` (candidate-side scoping only; inbound links counted from any source). Pinned by `test/sum-stale-chunk-chars.test.ts`, `test/embedding-signature-stale.test.ts`, `test/e2e/engine-parity.test.ts`. Free-text alias layer: `resolveAliases(aliasNorms, opts?): Promise>>` (READ; maps each normalized alias to declaring `(slug, source_id)` pairs, source-scoped) and `setPageAliases(slug, sourceId, aliasNorms)` (WRITE; replaces the full alias set, delete-then-insert, empty clears, idempotent on the unique triple), called by the `importFromContent` ingest projection and the `reindex --aliases` backfill; parity across both engines, Pinned by `test/search/page-aliases-engine.test.ts`. `searchVector` in both engines injects the shared `buildBestPerPagePoolCte` per-page max-pool so a page surfaces on its strongest chunk. `executeRawDirect(sql, params?, opts?)` is the lock-hot-path sibling of `executeRaw`: same single-statement contract, but routes to the direct session-mode pool when dual-pool is active (Postgres/Supabase port 5432) so a long-held lock heartbeat survives the transaction pooler's per-transaction connection recycling; PGLite delegates straight to `executeRaw` (no pooler). Both engines implement it; the Minion lock path (`claim`/`renewLock`) is the consumer. - `src/core/engine-constants.ts` — single source of truth for engine batch-sizing constants. Exports `DELETE_BATCH_SIZE = 500` consumed by both engines' `deletePages` + `resolveSlugsByPaths` and by the sync delete + rename loops. Lives outside `engine.ts` (the interface module) to avoid circular-import worry — bounded per-statement work for predictable lock hold time + write amplification. - `src/core/background-work.ts` (#1762/#1745/#1775) — process background-work registry: the single owner of "drain every fire-and-forget DB-write sink before the CLI disconnects." `registerBackgroundWorkDrainer({name, order, drain(timeoutMs), abort?})` + `drainAllBackgroundWorkForCliExit({timeoutMs})` over a `Map` (idempotent registration by name; `__registerDrainerForTest` returns an unregister handle). Drains in explicit `(order, name)` order — facts FIRST (order 0) so its abort-path DB `logIngest` runs against the freshest live engine — and AWAITS `abort()` only when `drain()` reports `unfinished>0`. Best-effort per drainer: one sink's failure never blocks the others or the disconnect. FOUR sinks register at module import (rule-of-four): `facts/queue.ts` (order 0; `abort`=`shutdown()` cancels a hung facts:absorb Haiku via internalAbort), `last-retrieved.ts` (order 1), `search/hybrid.ts` (order 2; `awaitPendingSearchCacheWrites` bounded via `Promise.race`), `eval-capture.ts` (order 3; `captureEvalCandidate` self-tracks its promise via `awaitPendingEvalCaptures`). Both CLI-exit paths (`src/cli.ts` op-dispatch finally + `handleCliOnly` finally) call it before `engine.disconnect()` — closing the PGLite busy-loop where `db.close()` raced an in-flight job and pinned the single-writer lock (#1762). CLI-EXIT-ONLY: the facts `shutdown()` abort is permanent process state, never call in a long-lived `gbrain serve`. Companion changes: `src/core/ai/gateway.ts` `withDefaultTimeout(caller, ms)` bounds every outbound AI call (chat 300s, embed+multimodal 60s; env `GBRAIN_AI_{CHAT,EMBED,MULTIMODAL}_TIMEOUT_MS`; composed with caller signals via `AbortSignal.any`) and the op-dispatch + handleCliOnly force-exit timers `process.exit(process.exitCode ?? 0)` so a hung disconnect can't mask an errored op as success; `src/core/postgres-engine.ts` `reconnect()` module-mode branch re-establishes via idempotent `db.connect()` + `connectionManager.setReadPool` refresh instead of `db.disconnect()` (no null window for concurrent ops; fail-loud on real connect failure — #1745); `src/core/search/hybrid.ts` `embedQueryBounded` + a shared `QueryEmbedDeadline` (6s, floored 2s per embed via `MIN_QUERY_EMBED_BUDGET_MS`; env `GBRAIN_QUERY_EMBED_TIMEOUT_MS`) bounds the cache-lookup AND inner query embeds so a stalled provider falls back to keyword instead of hanging past the 10s force-exit (#1775). Incorporates + hardens PR #1763 (@ElliotDrel). Pinned by `test/core/background-work.test.ts`, `test/search/query-embed-deadline.test.ts`, `test/eval-capture-drain.test.ts`, `test/e2e/postgres-reconnect-singleton.test.ts`, `test/e2e/pglite-cli-exit.serial.test.ts`, `test/fix-wave-structural.test.ts`. - `src/core/search/graph-signals.ts` — per-query graph-signals helper. `applyGraphSignals(results, engine, opts)` runs as the 4th post-fusion stage (after backlink/salience/recency). Three boosts: `ADJACENCY_BOOST=1.05` (page linked from 2+ OTHER top-K results — local hub for THIS query), `CROSS_SOURCE_BOOST=1.10` (page linked from 2+ DIFFERENT sources — corroborated across team brains, dormant in single-source brains), `SESSION_DEMOTE=0.95` (3+ results from same chat session — keep the highest-scoring at full score, demote the rest). All three inherit the floor-ratio gate preventing weak pages from being boosted past strong ones via popularity. `computeScoreDistribution(results)` emits min/p25/p50/p75/p95/max + `reorder_band_width`. `sessionPrefix(slug)` extracts the chat-session anchor (`chat/2026-05-15-...`). Pure `pairedBootstrapPValue(deltas, resamples, rng)` exported for eval gates. Test seam via `adjacencyFn` DI. Fail-open: any error logs via `logGraphSignalsFailure` (JSONL audit via `audit-writer`) and returns the input array unchanged. Pinned by `test/search/graph-signals.test.ts` (incl. the IRON-RULE floor-gate regression). @@ -209,7 +209,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/commands/lint.ts` + `src/commands/sources.ts` extensions — `gbrain lint` gains a `markup-heavy` rule (flags pages whose prose-vs-markup ratio exceeds `content_sanity.max_markup_ratio`, reusing `assessContentSanity` so lint/gate/scan share one assessor); pinned by `test/lint-content-sanity.test.ts`. `gbrain sources audit ` becomes disposition-aware: its dry-run disk scan reports would-quarantine / would-reject / would-flag counts driven by the effective `content_sanity.junk_disposition` + markup config, so an operator previews the gate's verdict before sync. The `content-sanity-audit` JSONL (`src/core/audit/content-sanity-audit.ts`) records the new quarantine/flag dispositions. - `src/core/zombie-reap.ts` — idempotent `installSigchldHandler()` so JS-spawned children get reaped via Bun's internal `waitpid()`. Bun (like Node) only auto-reaps when a SIGCHLD listener is registered; without it, every child the worker spawns (shell jobs, embed batches, sub-agents) becomes a zombie on exit and holds connection slots. Called once at module load from `src/cli.ts` (Windows platform guard — SIGCHLD doesn't exist on Windows). Cross-file leak guard via `_uninstallSigchldHandlerForTests()`. Layer 1 of the three-layer zombie defense; Layer 2 is tini-as-PID-1 wrapping the worker subtree (via `src/core/minions/spawn-helpers.ts`); Layer 3 is the container's own tini for hard Bun crashes. - `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/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. `claim` and `renewLock` issue their UPDATE via `engine.executeRawDirect` (not `executeRaw`) so the lock heartbeat runs on the direct session-mode pool that the transaction pooler won't recycle mid-hold; on PGLite this is identical to `executeRaw`. Guarded by `test/queue-lock-retry.test.ts` (claim never falls back to `executeRaw`) and `test/postgres-execute-raw-direct.test.ts` (routing decision matrix). - `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`. 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). diff --git a/package.json b/package.json index d9b60166f..9fd5009e6 100644 --- a/package.json +++ b/package.json @@ -143,5 +143,5 @@ "bun": ">=1.3.10" }, "license": "MIT", - "version": "0.42.23.0" + "version": "0.42.24.0" } diff --git a/src/core/engine.ts b/src/core/engine.ts index 2d739c2a7..4d52f3b3e 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -1912,6 +1912,24 @@ export interface BrainEngine { opts?: { signal?: AbortSignal }, ): Promise; + /** + * Like `executeRaw`, but routes through the DIRECT (session-mode) pool when + * dual-pool is active (Supabase: port 5432), falling back to the read pool + * otherwise. Use this for the Minion lock hot-path (`claim`/`renewLock`): + * those statements heartbeat a lock over many seconds, and the + * transaction-mode pooler (port 6543) recycles connections per-transaction, + * which surfaces as `CONNECTION_ENDED` mid-heartbeat → orphaned locks → + * silent worker wedge. The direct session pool holds the connection open for + * the life of the worker, so heartbeats survive. Single-statement UPDATEs + * only — same idempotency contract as `executeRaw`. On PGLite (no pooler) + * this is identical to `executeRaw`. + */ + executeRawDirect>( + sql: string, + params?: unknown[], + opts?: { signal?: AbortSignal }, + ): Promise; + // ============================================================ // v0.20.0 Cathedral II: code edges (Layer 5 populates, Layer 7 consumes) // ============================================================ diff --git a/src/core/minions/queue.ts b/src/core/minions/queue.ts index 8c1eef51f..8dd0aa85e 100644 --- a/src/core/minions/queue.ts +++ b/src/core/minions/queue.ts @@ -601,7 +601,10 @@ export class MinionQueue { async claim(lockToken: string, lockDurationMs: number, queue: string, registeredNames: string[]): Promise { if (registeredNames.length === 0) return null; - const rows = await this.engine.executeRaw>( + // Direct (session-mode) pool: claim opens the lock that renewLock then + // heartbeats. Both must live on a connection the transaction-mode pooler + // won't recycle mid-hold, or the lock orphans and the worker wedges. + const rows = await this.engine.executeRawDirect>( `UPDATE minion_jobs SET status = 'active', lock_token = $1, @@ -1073,7 +1076,10 @@ export class MinionQueue { /** Renew lock (token-fenced). Returns false if token mismatch (job was reclaimed). */ async renewLock(id: number, lockToken: string, lockDurationMs: number): Promise { - const rows = await this.engine.executeRaw>( + // Direct (session-mode) pool — see claim(). The heartbeat that keeps a job + // alive for minutes cannot run on the transaction pooler without periodic + // CONNECTION_ENDED drops that look like lock-expiry and orphan the job. + const rows = await this.engine.executeRawDirect>( `UPDATE minion_jobs SET lock_until = now() + ($1::double precision * interval '1 millisecond'), updated_at = now() WHERE id = $2 AND lock_token = $3 AND status = 'active' RETURNING id`, diff --git a/src/core/pglite-engine.ts b/src/core/pglite-engine.ts index 93157bb9a..5fb43ade1 100644 --- a/src/core/pglite-engine.ts +++ b/src/core/pglite-engine.ts @@ -4756,6 +4756,20 @@ export class PGLiteEngine implements BrainEngine { return Promise.race([queryPromise, abortPromise]); } + /** + * PGLite is in-process WASM with no connection pooler, so the direct-pool + * routing that `executeRawDirect` provides on Postgres is a no-op here: + * delegate straight to `executeRaw`. Present so the BrainEngine contract is + * satisfied and the Minion lock hot-path works identically on both engines. + */ + async executeRawDirect>( + sql: string, + params?: unknown[], + opts?: { signal?: AbortSignal }, + ): Promise { + return this.executeRaw(sql, params, opts); + } + // ============================================================ // v0.20.0 Cathedral II: code edges (Layer 1 stubs — filled by Layer 5) // ============================================================ diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 9c72f21b3..f1cc7a794 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -4879,17 +4879,25 @@ export class PostgresEngine implements BrainEngine { } } - async executeRaw>( + /** + * Shared body for executeRaw / executeRawDirect: run a raw statement on the + * given connection and wire AbortSignal cancellation onto the pending query. + * The ONLY difference between the two public methods is which connection they + * pick (read pool vs direct session pool), so the cancellation plumbing lives + * here in one place rather than being copy-pasted. + * + * v0.41.18.0 (A20, codex #7): real cancellation via postgres.js's .cancel() + * on the pending query. Init nudge (3s wallclock cap) is the first consumer; + * the AbortSignal fires when the timer trips. An already-aborted signal + * short-circuits before the network round-trip. + */ + private runUnsafe( + conn: ReturnType, sql: string, params?: unknown[], opts?: { signal?: AbortSignal }, ): Promise { - const conn = this.sql; const pending = conn.unsafe(sql, params as Parameters[1]); - // v0.41.18.0 (A20, codex #7): real cancellation via postgres.js's - // .cancel() on the pending query. Init nudge (3s wallclock cap) is the - // first consumer; the AbortSignal fires when the timer trips. - // Already-aborted signal short-circuits before the network round-trip. if (opts?.signal) { if (opts.signal.aborted) { // .cancel() is fire-and-forget; the awaited query rejects with the @@ -4905,7 +4913,7 @@ export class PostgresEngine implements BrainEngine { try { (pending as unknown as { cancel?: () => void }).cancel?.(); } catch { - // best-effort; the .then below settles regardless + // best-effort; the .finally below settles regardless } }; opts.signal.addEventListener('abort', onAbort, { once: true }); @@ -4913,7 +4921,15 @@ export class PostgresEngine implements BrainEngine { opts.signal?.removeEventListener('abort', onAbort); }); } - return pending as unknown as T[]; + return pending as unknown as Promise; + } + + async executeRaw>( + sql: string, + params?: unknown[], + opts?: { signal?: AbortSignal }, + ): Promise { + return this.runUnsafe(this.sql, sql, params, opts); // Pre-#406 behavior: throw on any error including connection death. // Per-call auto-retry is not safe here because executeRaw is also used // for non-transactional mutations (DELETE/UPDATE/INSERT in sources.ts, @@ -4924,6 +4940,34 @@ export class PostgresEngine implements BrainEngine { // swap in a fresh pool. See db.ts setSessionDefaults / supervisor.ts. } + /** + * Minion lock hot-path variant of executeRaw. Routes to the DIRECT + * session-mode pool (port 5432) when dual-pool is active so lock + * heartbeats survive the transaction-pooler's per-transaction connection + * recycling. See BrainEngine.executeRawDirect for the full rationale. + * + * When this engine is a transaction-scoped clone (txEngine from + * transaction()), `connectionManager` is inherited but `this.sql` is the tx + * connection; we intentionally honor the tx connection in that case by + * falling through to this.sql, because routing a statement inside an open + * transaction onto a different pool would break atomicity. The lock + * hot-path (claim/renewLock) does NOT run inside transaction(), so in + * practice this always reaches the direct pool there. + */ + async executeRawDirect>( + sql: string, + params?: unknown[], + opts?: { signal?: AbortSignal }, + ): Promise { + // Inside an open transaction, _sql is the reserved tx connection (set via + // defineProperty in transaction()); never reroute off it. + const inTransaction = this._sql !== null && this.connectionManager?.peekReadPool() !== this._sql; + const conn = (!inTransaction && this.connectionManager?.isDualPoolActive()) + ? await this.connectionManager.ddl() + : this.sql; + return this.runUnsafe(conn, sql, params, opts); + } + // ============================================================ // v0.20.0 Cathedral II: code edges (Layer 1 stubs — filled by Layer 5) // ============================================================ diff --git a/test/connection-resilience.test.ts b/test/connection-resilience.test.ts index 716e125b5..3f73d7eff 100644 --- a/test/connection-resilience.test.ts +++ b/test/connection-resilience.test.ts @@ -262,28 +262,46 @@ describe('Eng-review D3 — executeRaw has no per-call retry wrapper', () => { it('PostgresEngine.executeRaw is a single-statement passthrough (no try/catch on connection errors)', () => { const src = readFileSync(resolve('src/core/postgres-engine.ts'), 'utf-8'); - // Find the executeRaw method in the class (not the helper inside withReservedConnection) - // v0.41.18.0 (T5/A20): signature extended with optional `opts?: { signal?: AbortSignal }` - // 3rd arg + multi-line shape for real query cancellation. Regex updated to - // tolerate both the legacy single-line and the new multi-line signatures. + // v0.42.24.0 (eng-review D1): the cancellation plumbing shared by executeRaw + // and executeRawDirect was extracted into a private `runUnsafe(conn, ...)` + // helper. executeRaw / executeRawDirect now pick a connection and delegate; + // the single `conn.unsafe(` call lives in runUnsafe. The D3 invariant (no + // per-call retry wrapper) is unchanged — it just spans the delegate now, so + // this guard checks both the public methods AND the shared helper. + + // Find executeRaw in the class (not the helper inside withReservedConnection). + // v0.41.18.0 (T5/A20): signature extended with optional `opts?: { signal?: AbortSignal }`. const fnMatch = src.match(/async executeRaw>\(\s*sql: string,\s*params\?: unknown\[\][^)]*\):\s*Promise\s*\{([\s\S]*?)\n \}/); expect(fnMatch).not.toBeNull(); const body = fnMatch![1]; - // Must not call reconnect() from this method (D3 intent: no per-call - // retry — recovery is supervisor-driven via reconnect()). + // executeRaw must not retry: no reconnect, no inline re-issue, and it must + // delegate to runUnsafe rather than re-implementing the query path. expect(body).not.toContain('this.reconnect()'); - // Must call conn.unsafe directly, exactly ONCE (no retry re-issue). - expect(body).toContain('conn.unsafe('); - const unsafeCallCount = (body.match(/conn\.unsafe\(/g) || []).length; - expect(unsafeCallCount).toBe(1); - // The try/catch present here is ONLY for AbortSignal cancellation - // swallow (v0.41.18.0 A20), NOT for connection retry. Confirm by checking - // the swallowed throws are .cancel() not network re-issue. - if (body.includes('catch')) { + expect(body).toContain('this.runUnsafe'); + expect((body.match(/conn\.unsafe\(/g) || []).length).toBe(0); + + // executeRawDirect (the Minion lock hot-path sibling) routes to the direct + // session pool but must NOT introduce a retry wrapper either — same delegate. + const directMatch = src.match(/async executeRawDirect>\(\s*sql: string,\s*params\?: unknown\[\][^)]*\):\s*Promise\s*\{([\s\S]*?)\n \}/); + expect(directMatch).not.toBeNull(); + const directBody = directMatch![1]; + expect(directBody).not.toContain('this.reconnect()'); + expect(directBody).toContain('this.runUnsafe'); + expect((directBody.match(/conn\.unsafe\(/g) || []).length).toBe(0); + + // The shared helper issues conn.unsafe EXACTLY ONCE (no retry re-issue) and + // never reconnects. Its try/catch is ONLY the AbortSignal cancellation + // swallow (v0.41.18.0 A20), NOT a connection retry. + const helperMatch = src.match(/private runUnsafe\(\s*conn:[^)]*\):\s*Promise\s*\{([\s\S]*?)\n \}/); + expect(helperMatch).not.toBeNull(); + const helperBody = helperMatch![1]; + expect(helperBody).not.toContain('this.reconnect()'); + expect((helperBody.match(/conn\.unsafe\(/g) || []).length).toBe(1); + if (helperBody.includes('catch')) { // If catch exists, it must be the cancel-swallow shape, NOT a retry shape. - expect(body).not.toMatch(/catch[^{]*\{[\s\S]*?conn\.unsafe/); - expect(body).not.toMatch(/catch[^{]*\{[\s\S]*?setTimeout/); + expect(helperBody).not.toMatch(/catch[^{]*\{[\s\S]*?conn\.unsafe/); + expect(helperBody).not.toMatch(/catch[^{]*\{[\s\S]*?setTimeout/); } }); diff --git a/test/postgres-execute-raw-direct.test.ts b/test/postgres-execute-raw-direct.test.ts new file mode 100644 index 000000000..cd9a54458 --- /dev/null +++ b/test/postgres-execute-raw-direct.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, test } from 'bun:test'; +import { PostgresEngine } from '../src/core/postgres-engine.ts'; + +/** + * executeRawDirect routing decision (PR #1816 lock hot-path fix). + * + * The point of executeRawDirect is to send the Minion lock heartbeat + * (claim/renewLock) to the DIRECT session-mode pool (port 5432) instead of the + * transaction pooler (6543) that reaps connections mid-hold. That routing + * branch only fires against a real Supabase dual-pool, which CI doesn't have — + * so the decision itself (which connection gets the statement) went untested. + * + * This exercises the pure routing logic with a stubbed ConnectionManager and + * fake Sql handles, covering all three shapes: + * + * ┌─────────────────────────────┬──────────────┬───────────────┐ + * │ engine shape │ dual-pool │ conn chosen │ + * ├─────────────────────────────┼──────────────┼───────────────┤ + * │ worker, not in tx │ active │ ddl() direct │ + * │ tx clone (_sql = tx conn) │ active │ this.sql (tx) │ ← atomicity + * │ worker, not in tx │ inactive │ this.sql read │ + * └─────────────────────────────┴──────────────┴───────────────┘ + */ + +type FakeSql = { unsafe: (sql: string, params?: unknown[]) => Promise }; + +/** A fake postgres.js handle whose unsafe() tags rows with its label. */ +function fakeSql(label: string): FakeSql { + return { + unsafe: async () => [{ via: label }], + }; +} + +/** + * Build a PostgresEngine with its connection internals stubbed so + * executeRawDirect can be driven without a live database. + * + * - readConn → the read pool (what `this.sql` returns when _sql === readPool) + * - directConn → what connectionManager.ddl() resolves to + * - sqlOverride → forces the `get sql()` getter (used to model a tx clone whose + * `this.sql` is the tx connection, distinct from peekReadPool()). + */ +function makeEngine(opts: { + dualPoolActive: boolean; + readConn: FakeSql; + directConn: FakeSql; + // When set, models a tx clone: _sql is the tx conn (!== peekReadPool()). + txConn?: FakeSql; +}): PostgresEngine { + const engine = new PostgresEngine(); + const e = engine as unknown as Record; + + // _sql: tx conn for a clone, otherwise the read pool itself (the worker case + // where connect() does setReadPool(this._sql)). + e._sql = opts.txConn ?? opts.readConn; + + // The `get sql()` getter on the prototype returns _sql when set, so we don't + // need to override it — _sql already drives it. For a tx clone _sql is txConn, + // so this.sql === txConn, exactly as Object.defineProperty does in transaction(). + + e.connectionManager = { + isDualPoolActive: () => opts.dualPoolActive, + peekReadPool: () => opts.readConn, + ddl: async () => opts.directConn, + }; + + return engine; +} + +describe('PostgresEngine.executeRawDirect — routing decision (PR #1816)', () => { + test('dual-pool active + not in tx → routes to direct (ddl) pool', async () => { + const readConn = fakeSql('read'); + const directConn = fakeSql('direct'); + const engine = makeEngine({ dualPoolActive: true, readConn, directConn }); + + const rows = await engine.executeRawDirect<{ via: string }>('UPDATE minion_jobs SET x=1'); + expect(rows[0].via).toBe('direct'); + }); + + test('inside a transaction → honors the tx connection (never reroutes off it)', async () => { + const readConn = fakeSql('read'); + const directConn = fakeSql('direct'); + const txConn = fakeSql('tx'); + // dual-pool is active, but _sql (tx) !== peekReadPool() (read) → inTransaction. + const engine = makeEngine({ dualPoolActive: true, readConn, directConn, txConn }); + + const rows = await engine.executeRawDirect<{ via: string }>('UPDATE minion_jobs SET x=1'); + expect(rows[0].via).toBe('tx'); + }); + + test('dual-pool inactive → falls back to the read pool', async () => { + const readConn = fakeSql('read'); + const directConn = fakeSql('direct'); + const engine = makeEngine({ dualPoolActive: false, readConn, directConn }); + + const rows = await engine.executeRawDirect<{ via: string }>('UPDATE minion_jobs SET x=1'); + expect(rows[0].via).toBe('read'); + }); + + test('already-aborted signal short-circuits with AbortError before routing the query', async () => { + const readConn = fakeSql('read'); + const directConn = fakeSql('direct'); + const engine = makeEngine({ dualPoolActive: true, readConn, directConn }); + + const ac = new AbortController(); + ac.abort(); + await expect( + engine.executeRawDirect('UPDATE minion_jobs SET x=1', [], { signal: ac.signal }), + ).rejects.toThrow(/abort/i); + }); +}); diff --git a/test/queue-lock-retry.test.ts b/test/queue-lock-retry.test.ts index f881589ad..14bc9c082 100644 --- a/test/queue-lock-retry.test.ts +++ b/test/queue-lock-retry.test.ts @@ -58,7 +58,11 @@ describe('MinionQueue lock-path recovery (issue #1678)', () => { let calls = 0; const engine = { kind: 'postgres', - executeRaw: async () => { calls++; throw connEndedError(); }, + // claim() routes through executeRawDirect (direct session pool) as of + // the lock-hot-path fix; executeRaw is kept as a throwing guard to + // prove claim never falls back to it. + executeRawDirect: async () => { calls++; throw connEndedError(); }, + executeRaw: async () => { throw new Error('claim must not use executeRaw'); }, reconnect: async () => {}, } as unknown as ConstructorParameters[0];